MARGIN_WHOLESALE.sql

Source: DMF/PuC/AIF/DR3/MARGIN_WHOLESALE.sql

-- ============================================================================
-- IC MARGIN WHOLESALE — Full Validation Pipeline
-- ============================================================================
-- Fact:           ic_margin_wholesale
-- Source:         ORI_MARGIN_WHOLESALE_RAW
-- BQ source:      `puc-p-dataf-common`.export_oracle_migration.margin_history_wholesale
-- Rejected:       ORI_MARGIN_WHOLESALE_REJECTED
-- STG (accepted): MARGIN_HISTORY_WHOLESALE_STG
-- Rules:          13
-- Pipeline:       RAW → REJECTED → STG → CTRL
--
-- How to run in SQL Developer:
--   Open this file → F5 (Run as Script)
--   Each step can also be highlighted and run independently with F9.
--
-- Prerequisites:
--   XREF_ITEM_IBC_RETAIL  built by ../ItemPR1_PV1.sql
--
-- Collapsed from phase1/phase2/phase3 playbooks generated 2026-06-07T17:56:07.
-- Phase 3 contained phases 1 and 2 verbatim; the per-rule DIAGNOSTIC
-- ALTERNATIVE block duplicated the unified INSERT above it and was removed.
-- ============================================================================
 
ALTER SESSION SET DDL_LOCK_TIMEOUT = 300;
 
-- ============================================================================
-- STEP 1 — Rebuild validity materialization tables (run once / when stale)
-- ============================================================================
-- The framework caches these tables across runs (the underlying
-- master-data joins are expensive but the master data changes rarely).
-- ONLY run this block when:
--   • first build on a fresh schema,
--   • the [dimensions.<name>] validity_query was edited,
--   • the master-data tables were reloaded.
 
-- Dimension 'item_valid' → DMF_VALIDITY__ITEM_VALID
DROP TABLE DMF_VALIDITY__ITEM_VALID PURGE;  -- ignore ORA-00942 on first run
CREATE TABLE DMF_VALIDITY__ITEM_VALID AS
SELECT DISTINCT imc.item AS item
  FROM item_master_ctrl imc
 WHERE imc.ctrl_status = 'A'
UNION
SELECT DISTINCT xi.item_pr1 AS item
  FROM item_master_ctrl imc
  JOIN XREF_ITEM_IBC_RETAIL xi ON xi.item_pv1 = imc.item
 WHERE imc.ctrl_status = 'A'
   AND xi.item_pr1 IS NOT NULL;
CREATE INDEX IX_DMF_VALIDITY__ITEM_VALID ON DMF_VALIDITY__ITEM_VALID (ITEM);
 
-- Dimension 'organization_wh_only' → DMF_VALIDITY__ORGANIZAT_48D17B
DROP TABLE DMF_VALIDITY__ORGANIZAT_48D17B PURGE;  -- ignore ORA-00942 on first run
CREATE TABLE DMF_VALIDITY__ORGANIZAT_48D17B AS
SELECT TO_CHAR(STORE) AS org_num FROM STORE_ADD_CTRL WHERE ctrl_status = 'C'
UNION ALL
SELECT TO_CHAR(WH)    AS org_num FROM WH_CTRL         WHERE ctrl_status = 'C';
CREATE INDEX IX_DMF_VALIDITY__ORGANIZAT_48D ON DMF_VALIDITY__ORGANIZAT_48D17B (ORG_NUM);
 
-- Verify the validity tables populated:
SELECT 'DMF_VALIDITY__ITEM_VALID' AS dim, COUNT(*) FROM DMF_VALIDITY__ITEM_VALID;
SELECT 'DMF_VALIDITY__ORGANIZAT_48D17B' AS dim, COUNT(*) FROM DMF_VALIDITY__ORGANIZAT_48D17B;
 
-- ============================================================================
-- STEP 2 — Materialize normalized source into DMF_SOURCE__IC_MARGIN_WHOLESAL
-- ============================================================================
-- One scan of RAW now → all subsequent rules read from a smaller, hotter,
-- indexed snapshot. Source CTAS override applied (column renames / JOINs).
DROP TABLE DMF_SOURCE__IC_MARGIN_WHOLESAL PURGE;  -- ignore ORA-00942 on first run
CREATE TABLE DMF_SOURCE__IC_MARGIN_WHOLESAL PARALLEL 4 AS
SELECT /*+ PARALLEL(4) */
  ROWID                                   AS SRC_ROWID,
  r.ITEM,
  LTRIM(r.ORG_NUM, '0')                  AS ORG_NUM,
  r.DAY_DT,
  r.ICM_QTY,
  r.ICM_COST_AMT_LCL,
  CAST(NULL AS NUMBER(20,4))              AS ICM_RTL_AMT_LCL,
  CAST(NULL AS NUMBER(22,7))              AS GLOBAL1_EXCHANGE_RATE,
  CAST(NULL AS NUMBER(22,7))              AS GLOBAL2_EXCHANGE_RATE,
  CAST(NULL AS NUMBER(22,7))              AS GLOBAL3_EXCHANGE_RATE,
  r.LOC_CURR_CODE,
  CAST(NULL AS NUMBER(22,7))              AS LOC_EXCHANGE_RATE,
  r.DOC_CURR_CODE,
  CAST(NULL AS NUMBER(4,0))               AS ETL_THREAD_VAL,
  CAST(NULL AS CHAR(1))                   AS DELETE_FLG,
  CAST(NULL AS NUMBER(10,0))              AS DATASOURCE_NUM_ID
FROM ORI_MARGIN_WHOLESALE_RAW r;
 
-- Index on the row-identity tuple makes duplicate_key and the STG
-- anti-join cheap.
CREATE INDEX IX_DMF_SOURCE__IC_MARGIN_WHO_RID ON DMF_SOURCE__IC_MARGIN_WHOLESAL (ITEM, ORG_NUM, DAY_DT);
 
-- Reset parallel attribute so per-rule reads don't auto-parallelize.
ALTER TABLE DMF_SOURCE__IC_MARGIN_WHOLESAL NOPARALLEL;
 
-- Sanity check:
SELECT COUNT(*) FROM DMF_SOURCE__IC_MARGIN_WHOLESAL;
 
-- ============================================================================
-- STEP 3 — Materialize duplicate-key set into DMF_DUPKEYS__IC_MARGIN_WHOLESA
-- ============================================================================
-- One GROUP BY HAVING over the source now → the duplicate_key rule
-- becomes a cheap semi-join against this small indexed table.
DROP TABLE DMF_DUPKEYS__IC_MARGIN_WHOLESA PURGE;  -- ignore ORA-00942 on first run
CREATE TABLE DMF_DUPKEYS__IC_MARGIN_WHOLESA AS
SELECT ITEM, ORG_NUM, DAY_DT FROM DMF_SOURCE__IC_MARGIN_WHOLESAL
 GROUP BY ITEM, ORG_NUM, DAY_DT
HAVING COUNT(*) > 1;
 
CREATE INDEX IX_DMF_DUPKEYS__IC_MARGIN_WH_KEY ON DMF_DUPKEYS__IC_MARGIN_WHOLESA (ITEM, ORG_NUM, DAY_DT);
 
-- Sanity check (size of the conflict set):
SELECT COUNT(*) FROM DMF_DUPKEYS__IC_MARGIN_WHOLESA;
 
-- ============================================================================
-- STEP 4 — Ensure ORI_MARGIN_WHOLESALE_REJECTED exists, then empty it
-- ============================================================================
BEGIN
  EXECUTE IMMEDIATE 'CREATE TABLE ORI_MARGIN_WHOLESALE_REJECTED (     LOG_LVL VARCHAR2(1),     ERROR_MSG VARCHAR2(255),     VALIDATION_RUN_ID VARCHAR2(80),     RULE_ID VARCHAR2(120),     SRC_ROWID VARCHAR2(64),     ITEM VARCHAR2(4000),     ORG_NUM VARCHAR2(4000),     DAY_DT VARCHAR2(4000),     ICM_QTY VARCHAR2(4000),     ICM_COST_AMT_LCL VARCHAR2(4000),     ICM_RTL_AMT_LCL VARCHAR2(4000),     GLOBAL1_EXCHANGE_RATE VARCHAR2(4000),     GLOBAL2_EXCHANGE_RATE VARCHAR2(4000),     GLOBAL3_EXCHANGE_RATE VARCHAR2(4000),     LOC_CURR_CODE VARCHAR2(4000),     LOC_EXCHANGE_RATE VARCHAR2(4000),     DOC_CURR_CODE VARCHAR2(4000),     ETL_THREAD_VAL VARCHAR2(4000),     DELETE_FLG VARCHAR2(4000),     DATASOURCE_NUM_ID VARCHAR2(4000),     INSERTED_AT TIMESTAMP(6) )';
EXCEPTION WHEN OTHERS THEN
  IF SQLCODE != -955 THEN RAISE; END IF;  -- -955 = table already exists
END;
/
TRUNCATE TABLE ORI_MARGIN_WHOLESALE_REJECTED;
-- If TRUNCATE fails with ORA-00054 (orphan lock), use DELETE instead:
-- DELETE FROM ORI_MARGIN_WHOLESALE_REJECTED;
-- COMMIT;
 
-- ============================================================================
-- STEP 5 — Unified UNION ALL INSERT (single source scan)
-- ============================================================================
-- One INSERT covers every rule via UNION ALL. With the source CTE
-- materialized into DMF_SOURCE__<fact>, Oracle scans the snapshot once
-- and evaluates each rule's predicate against the buffered table.
INSERT INTO ORI_MARGIN_WHOLESALE_REJECTED (
  LOG_LVL, ERROR_MSG, VALIDATION_RUN_ID, RULE_ID, SRC_ROWID, ITEM, ORG_NUM, DAY_DT, ICM_QTY, ICM_COST_AMT_LCL, ICM_RTL_AMT_LCL, GLOBAL1_EXCHANGE_RATE, GLOBAL2_EXCHANGE_RATE, GLOBAL3_EXCHANGE_RATE, LOC_CURR_CODE, LOC_EXCHANGE_RATE, DOC_CURR_CODE, ETL_THREAD_VAL, DELETE_FLG, DATASOURCE_NUM_ID, INSERTED_AT
)
WITH s AS (
    SELECT
      SRC_ROWID,
      ITEM,
      ORG_NUM,
      DAY_DT,
      ICM_QTY,
      ICM_COST_AMT_LCL,
      ICM_RTL_AMT_LCL,
      GLOBAL1_EXCHANGE_RATE,
      GLOBAL2_EXCHANGE_RATE,
      GLOBAL3_EXCHANGE_RATE,
      LOC_CURR_CODE,
      LOC_EXCHANGE_RATE,
      DOC_CURR_CODE,
      ETL_THREAD_VAL,
      DELETE_FLG,
      DATASOURCE_NUM_ID
    FROM DMF_SOURCE__IC_MARGIN_WHOLESAL
)
SELECT 'E', 'INVALID FORMAT: ITEM', 'MANUAL_RUN', 'ERROR_DTYPE_ITEM', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ITEM IS NOT NULL AND LENGTHB(TO_CHAR(ITEM)) > 80
UNION ALL
SELECT 'E', 'INVALID FORMAT: ORG_NUM', 'MANUAL_RUN', 'ERROR_DTYPE_ORG_NUM', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ORG_NUM IS NOT NULL AND LENGTHB(TO_CHAR(ORG_NUM)) > 80
UNION ALL
SELECT 'E', 'INVALID FORMAT: DAY_DT', 'MANUAL_RUN', 'ERROR_DTYPE_DAY_DT', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE DAY_DT IS NOT NULL AND TRUNC(DAY_DT) <> DAY_DT
UNION ALL
SELECT 'E', 'INVALID FORMAT: ICM_QTY', 'MANUAL_RUN', 'ERROR_DTYPE_ICM_QTY', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ICM_QTY IS NOT NULL AND (ABS(TRUNC(TO_NUMBER(ICM_QTY))) >= POWER(10,14) OR TO_NUMBER(ICM_QTY) != ROUND(TO_NUMBER(ICM_QTY),4))
UNION ALL
SELECT 'E', 'INVALID FORMAT: ICM_COST_AMT_LCL', 'MANUAL_RUN', 'ERROR_DTYPE_ICM_COST_AMT_LCL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ICM_COST_AMT_LCL IS NOT NULL AND (ABS(TRUNC(TO_NUMBER(ICM_COST_AMT_LCL))) >= POWER(10,16) OR TO_NUMBER(ICM_COST_AMT_LCL) != ROUND(TO_NUMBER(ICM_COST_AMT_LCL),4))
UNION ALL
SELECT 'E', 'INVALID FORMAT: LOC_CURR_CODE', 'MANUAL_RUN', 'ERROR_DTYPE_LOC_CURR_CODE', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE LOC_CURR_CODE IS NOT NULL AND LENGTHB(TO_CHAR(LOC_CURR_CODE)) > 30
UNION ALL
SELECT 'E', 'INVALID FORMAT: DOC_CURR_CODE', 'MANUAL_RUN', 'ERROR_DTYPE_DOC_CURR_CODE', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE DOC_CURR_CODE IS NOT NULL AND LENGTHB(TO_CHAR(DOC_CURR_CODE)) > 30
UNION ALL
SELECT 'E', 'ITEM IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ITEM IS NULL
UNION ALL
SELECT 'E', 'ORG_NUM IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ORG_NUM IS NULL
UNION ALL
SELECT 'E', 'DAY_DT IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE DAY_DT IS NULL
UNION ALL
SELECT 'E', 'ITEM NOT FOUND: ITEM_MASTER_CTRL', 'MANUAL_RUN', 'ERROR_MISSING_ITEM', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ITEM IS NOT NULL AND NOT EXISTS (SELECT 1 FROM DMF_VALIDITY__ITEM_VALID d WHERE d.ITEM = s.ITEM)
UNION ALL
SELECT 'E', 'ORG_NUM NOT FOUND: WH_CTRL', 'MANUAL_RUN', 'ERROR_MISSING_ORG_NUM', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ORG_NUM IS NOT NULL AND NOT EXISTS (SELECT 1 FROM DMF_VALIDITY__ORGANIZAT_48D17B d WHERE d.ORG_NUM = s.ORG_NUM)
UNION ALL
SELECT 'W', 'WARN_NEGATIVE_ICM_QTY', 'MANUAL_RUN', 'WARN_NEGATIVE_ICM_QTY', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ICM_QTY IS NOT NULL AND ICM_QTY < 0
UNION ALL
SELECT 'W', 'WARN_NEGATIVE_ICM_COST', 'MANUAL_RUN', 'WARN_NEGATIVE_ICM_COST', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE ICM_COST_AMT_LCL IS NOT NULL AND ICM_COST_AMT_LCL < 0
UNION ALL
SELECT 'E', 'DUPLICATE KEY: ITEM, ORG_NUM, DAY_DT', 'MANUAL_RUN', 'ERR_DUP_KEY_CONFLICT', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE FROM s WHERE (ITEM, ORG_NUM, DAY_DT) IN (SELECT ITEM, ORG_NUM, DAY_DT FROM DMF_DUPKEYS__IC_MARGIN_WHOLESA);
COMMIT;
 
-- ============================================================================
-- STEP 6 — Build typed MARGIN_HISTORY_WHOLESALE_STG (drop-and-recreate)
-- ============================================================================
-- DDL types are pulled from the RAP spec (not VARCHAR2(4000) blanket).
DROP TABLE MARGIN_HISTORY_WHOLESALE_STG PURGE;  -- ignore ORA-00942
CREATE TABLE MARGIN_HISTORY_WHOLESALE_STG (
    ITEM VARCHAR2(80 CHAR),
    ORG_NUM VARCHAR2(80 CHAR),
    DAY_DT DATE,
    ICM_QTY NUMBER(18,4),
    ICM_COST_AMT_LCL NUMBER(20,4),
    ICM_RTL_AMT_LCL NUMBER(20,4),
    GLOBAL1_EXCHANGE_RATE NUMBER(22,7),
    GLOBAL2_EXCHANGE_RATE NUMBER(22,7),
    GLOBAL3_EXCHANGE_RATE NUMBER(22,7),
    LOC_CURR_CODE VARCHAR2(30 CHAR),
    LOC_EXCHANGE_RATE NUMBER(22,7),
    DOC_CURR_CODE VARCHAR2(30 CHAR),
    ETL_THREAD_VAL NUMBER(4,0),
    DELETE_FLG CHAR(1 CHAR),
    DATASOURCE_NUM_ID NUMBER(10,0),
    INSERTED_AT TIMESTAMP(6)
);
 
-- ============================================================================
-- STEP 7 — Populate MARGIN_HISTORY_WHOLESALE_STG (anti-join on SRC_ROWID; LOG_LVL='W' rows still flow through)
-- ============================================================================
INSERT INTO MARGIN_HISTORY_WHOLESALE_STG (ITEM, ORG_NUM, DAY_DT, ICM_QTY, ICM_COST_AMT_LCL, ICM_RTL_AMT_LCL, GLOBAL1_EXCHANGE_RATE, GLOBAL2_EXCHANGE_RATE, GLOBAL3_EXCHANGE_RATE, LOC_CURR_CODE, LOC_EXCHANGE_RATE, DOC_CURR_CODE, ETL_THREAD_VAL, DELETE_FLG, DATASOURCE_NUM_ID, INSERTED_AT)
WITH s AS (
    SELECT
      SRC_ROWID,
      ITEM,
      ORG_NUM,
      DAY_DT,
      ICM_QTY,
      ICM_COST_AMT_LCL,
      ICM_RTL_AMT_LCL,
      GLOBAL1_EXCHANGE_RATE,
      GLOBAL2_EXCHANGE_RATE,
      GLOBAL3_EXCHANGE_RATE,
      LOC_CURR_CODE,
      LOC_EXCHANGE_RATE,
      DOC_CURR_CODE,
      ETL_THREAD_VAL,
      DELETE_FLG,
      DATASOURCE_NUM_ID
    FROM DMF_SOURCE__IC_MARGIN_WHOLESAL
)
SELECT s.ITEM, s.ORG_NUM, s.DAY_DT, s.ICM_QTY, s.ICM_COST_AMT_LCL, s.ICM_RTL_AMT_LCL, s.GLOBAL1_EXCHANGE_RATE, s.GLOBAL2_EXCHANGE_RATE, s.GLOBAL3_EXCHANGE_RATE, s.LOC_CURR_CODE, s.LOC_EXCHANGE_RATE, s.DOC_CURR_CODE, s.ETL_THREAD_VAL, s.DELETE_FLG, s.DATASOURCE_NUM_ID, SYSDATE
FROM s
WHERE NOT EXISTS (
  SELECT 1
  FROM ORI_MARGIN_WHOLESALE_REJECTED r
  WHERE r.VALIDATION_RUN_ID = 'MANUAL_RUN'
    AND r.LOG_LVL = 'E'
    AND r.SRC_ROWID = s.SRC_ROWID
)
  AND s.ITEM IS NOT NULL
  AND s.ORG_NUM IS NOT NULL
  AND s.DAY_DT IS NOT NULL;
COMMIT;
-- Row count check: should be COUNT(ORI_MARGIN_WHOLESALE_RAW) minus distinct SRC_ROWIDs in ORI_MARGIN_WHOLESALE_REJECTED where LOG_LVL='E'.
SELECT COUNT(*) AS stg_rows FROM MARGIN_HISTORY_WHOLESALE_STG;
 
-- ============================================================================
-- STEP 8 — Build MARGIN_HISTORY_WHOLESALE_CTRL (drop-and-recreate)
-- ============================================================================
-- Final Oracle export shape. Column projection comes from
-- `[ctrl_projection]` in the override; default = identity from STG.
DROP TABLE MARGIN_HISTORY_WHOLESALE_CTRL PURGE;  -- ignore ORA-00942
CREATE TABLE MARGIN_HISTORY_WHOLESALE_CTRL (
    ITEM VARCHAR2(80 CHAR),
    ORG_NUM VARCHAR2(80 CHAR),
    DAY_DT DATE,
    ICM_QTY NUMBER(18,4),
    ICM_COST_AMT_LCL NUMBER(20,4),
    ICM_RTL_AMT_LCL NUMBER(20,4),
    GLOBAL1_EXCHANGE_RATE NUMBER(22,7),
    GLOBAL2_EXCHANGE_RATE NUMBER(22,7),
    GLOBAL3_EXCHANGE_RATE NUMBER(22,7),
    LOC_CURR_CODE VARCHAR2(30 CHAR),
    LOC_EXCHANGE_RATE NUMBER(22,7),
    DOC_CURR_CODE VARCHAR2(30 CHAR),
    ETL_THREAD_VAL NUMBER(4,0),
    DELETE_FLG CHAR(1 CHAR),
    DATASOURCE_NUM_ID NUMBER(10,0),
    INSERTED_AT TIMESTAMP(6)
);
 
-- ============================================================================
-- STEP 9 — Populate MARGIN_HISTORY_WHOLESALE_CTRL from MARGIN_HISTORY_WHOLESALE_STG
-- ============================================================================
INSERT INTO MARGIN_HISTORY_WHOLESALE_CTRL (ITEM, ORG_NUM, DAY_DT, ICM_QTY, ICM_COST_AMT_LCL, ICM_RTL_AMT_LCL, GLOBAL1_EXCHANGE_RATE, GLOBAL2_EXCHANGE_RATE, GLOBAL3_EXCHANGE_RATE, LOC_CURR_CODE, LOC_EXCHANGE_RATE, DOC_CURR_CODE, ETL_THREAD_VAL, DELETE_FLG, DATASOURCE_NUM_ID, INSERTED_AT)
SELECT s.ITEM, s.ORG_NUM, s.DAY_DT, CASE WHEN s.ICM_QTY IS NULL OR ABS(TRUNC(s.ICM_QTY)) < 100000000000000 THEN s.ICM_QTY ELSE NULL END, CASE WHEN s.ICM_COST_AMT_LCL IS NULL OR ABS(TRUNC(s.ICM_COST_AMT_LCL)) < 10000000000000000 THEN s.ICM_COST_AMT_LCL ELSE NULL END, CASE WHEN s.ICM_RTL_AMT_LCL IS NULL OR ABS(TRUNC(s.ICM_RTL_AMT_LCL)) < 10000000000000000 THEN s.ICM_RTL_AMT_LCL ELSE NULL END, CASE WHEN s.GLOBAL1_EXCHANGE_RATE IS NULL OR ABS(TRUNC(s.GLOBAL1_EXCHANGE_RATE)) < 1000000000000000 THEN s.GLOBAL1_EXCHANGE_RATE ELSE NULL END, CASE WHEN s.GLOBAL2_EXCHANGE_RATE IS NULL OR ABS(TRUNC(s.GLOBAL2_EXCHANGE_RATE)) < 1000000000000000 THEN s.GLOBAL2_EXCHANGE_RATE ELSE NULL END, CASE WHEN s.GLOBAL3_EXCHANGE_RATE IS NULL OR ABS(TRUNC(s.GLOBAL3_EXCHANGE_RATE)) < 1000000000000000 THEN s.GLOBAL3_EXCHANGE_RATE ELSE NULL END, s.LOC_CURR_CODE, CASE WHEN s.LOC_EXCHANGE_RATE IS NULL OR ABS(TRUNC(s.LOC_EXCHANGE_RATE)) < 1000000000000000 THEN s.LOC_EXCHANGE_RATE ELSE NULL END, s.DOC_CURR_CODE, CASE WHEN s.ETL_THREAD_VAL IS NULL OR ABS(TRUNC(s.ETL_THREAD_VAL)) < 10000 THEN s.ETL_THREAD_VAL ELSE NULL END, s.DELETE_FLG, CASE WHEN s.DATASOURCE_NUM_ID IS NULL OR ABS(TRUNC(s.DATASOURCE_NUM_ID)) < 10000000000 THEN s.DATASOURCE_NUM_ID ELSE NULL END, SYSDATE
FROM MARGIN_HISTORY_WHOLESALE_STG s;
COMMIT;
SELECT COUNT(*) AS ctrl_rows FROM MARGIN_HISTORY_WHOLESALE_CTRL;
 
-- ============================================================================
-- STEP 10 — Rejection summary by rule
-- ============================================================================
SELECT RULE_ID, LOG_LVL, COUNT(*) AS rejected_rows
  FROM ORI_MARGIN_WHOLESALE_REJECTED
 GROUP BY RULE_ID, LOG_LVL
 ORDER BY rejected_rows DESC;