Source: DMF/PuC/AIF/FelipeBkp/price_init_delta.sql
-- ============================================================================
-- PRICE INIT DELTA — Full Validation Pipeline
-- ============================================================================
-- Fact: price_init_delta
-- Source: INIT_PRICE_DELTA_RAW
-- Rejected: INIT_PRICE_DELTA_REJECTED
-- STG (accepted): INIT_PRICE_DELTA_STG
-- Rules: 7
-- 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.
--
-- Collapsed from phase1/phase2/phase3 playbooks generated 2026-06-10T20:18:49.
-- 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 'organization_wh_direct' → DMF_VALIDITY__ORGANIZAT_A4FC2D
DROP TABLE DMF_VALIDITY__ORGANIZAT_A4FC2D PURGE; -- ignore ORA-00942 on first run
CREATE TABLE DMF_VALIDITY__ORGANIZAT_A4FC2D 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_A4F ON DMF_VALIDITY__ORGANIZAT_A4FC2D (ORG_NUM);
-- Verify the validity tables populated:
SELECT 'DMF_VALIDITY__ORGANIZAT_A4FC2D' AS dim, COUNT(*) FROM DMF_VALIDITY__ORGANIZAT_A4FC2D;
-- ============================================================================
-- STEP 2 — Materialize normalized source into DMF_SOURCE__PRICE_INIT_DELTA
-- ============================================================================
-- 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__PRICE_INIT_DELTA PURGE; -- ignore ORA-00942 on first run
CREATE TABLE DMF_SOURCE__PRICE_INIT_DELTA PARALLEL 4 AS
SELECT /*+ PARALLEL(4) */
ROWID AS SRC_ROWID,
r.ITEM,
TO_CHAR(r.STORE) AS ORG_NUM,
r.START_DATE AS DAY_DT,
r.PRICE_CHANGE_TRAN_TYPE,
r.PRICE AS STANDARD_UNIT_RTL_AMT_LCL,
r.PRICE AS SELLING_UNIT_RTL_AMT_LCL,
r.BASE_COST AS BASE_COST_AMT_LCL,
'EUR' AS LOC_CURR_CODE,
r.DOC_CURR_CODE
FROM INIT_PRICE_DELTA_RAW r;
-- Index on the row-identity tuple makes duplicate_key and the STG
-- anti-join cheap.
CREATE INDEX IX_DMF_SOURCE__PRICE_INIT_DE_RID ON DMF_SOURCE__PRICE_INIT_DELTA (ITEM, ORG_NUM, DAY_DT);
-- Reset parallel attribute so per-rule reads don't auto-parallelize.
ALTER TABLE DMF_SOURCE__PRICE_INIT_DELTA NOPARALLEL;
-- Sanity check:
SELECT COUNT(*) FROM DMF_SOURCE__PRICE_INIT_DELTA;
-- ============================================================================
-- STEP 3 — Materialize duplicate-key set into DMF_DUPKEYS__PRICE_INIT_DELTA
-- ============================================================================
-- 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__PRICE_INIT_DELTA PURGE; -- ignore ORA-00942 on first run
CREATE TABLE DMF_DUPKEYS__PRICE_INIT_DELTA AS
SELECT ITEM, ORG_NUM, DAY_DT FROM DMF_SOURCE__PRICE_INIT_DELTA
GROUP BY ITEM, ORG_NUM, DAY_DT
HAVING COUNT(*) > 1;
CREATE INDEX IX_DMF_DUPKEYS__PRICE_INIT_D_KEY ON DMF_DUPKEYS__PRICE_INIT_DELTA (ITEM, ORG_NUM, DAY_DT);
-- Sanity check (size of the conflict set):
SELECT COUNT(*) FROM DMF_DUPKEYS__PRICE_INIT_DELTA;
-- ============================================================================
-- STEP 4 — Ensure INIT_PRICE_DELTA_REJECTED exists, then empty it
-- ============================================================================
BEGIN
EXECUTE IMMEDIATE 'CREATE TABLE INIT_PRICE_DELTA_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), PRICE_CHANGE_TRAN_TYPE VARCHAR2(4000), STANDARD_UNIT_RTL_AMT_LCL VARCHAR2(4000), SELLING_UNIT_RTL_AMT_LCL VARCHAR2(4000), BASE_COST_AMT_LCL VARCHAR2(4000), LOC_CURR_CODE VARCHAR2(4000), DOC_CURR_CODE VARCHAR2(4000), INSERTED_AT TIMESTAMP(6) )';
EXCEPTION WHEN OTHERS THEN
IF SQLCODE != -955 THEN RAISE; END IF; -- -955 = table already exists
END;
/
TRUNCATE TABLE INIT_PRICE_DELTA_REJECTED;
-- If TRUNCATE fails with ORA-00054 (orphan lock), use DELETE instead:
-- DELETE FROM INIT_PRICE_DELTA_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 INIT_PRICE_DELTA_REJECTED (
LOG_LVL, ERROR_MSG, VALIDATION_RUN_ID, RULE_ID, SRC_ROWID, ITEM, ORG_NUM, DAY_DT, PRICE_CHANGE_TRAN_TYPE, STANDARD_UNIT_RTL_AMT_LCL, SELLING_UNIT_RTL_AMT_LCL, BASE_COST_AMT_LCL, LOC_CURR_CODE, DOC_CURR_CODE, INSERTED_AT
)
WITH s AS (
SELECT
SRC_ROWID,
ITEM,
ORG_NUM,
DAY_DT,
PRICE_CHANGE_TRAN_TYPE,
STANDARD_UNIT_RTL_AMT_LCL,
SELLING_UNIT_RTL_AMT_LCL,
BASE_COST_AMT_LCL,
LOC_CURR_CODE,
DOC_CURR_CODE
FROM DMF_SOURCE__PRICE_INIT_DELTA
)
SELECT 'E', 'ITEM IS NULL', '20260610T201849_7ba6ad21', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE ITEM IS NULL
UNION ALL
SELECT 'E', 'ORG_NUM IS NULL', '20260610T201849_7ba6ad21', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE ORG_NUM IS NULL
UNION ALL
SELECT 'E', 'DAY_DT IS NULL', '20260610T201849_7ba6ad21', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE DAY_DT IS NULL
UNION ALL
SELECT 'E', 'PRICE_CHANGE_TRAN_TYPE IS NULL', '20260610T201849_7ba6ad21', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE PRICE_CHANGE_TRAN_TYPE IS NULL
UNION ALL
SELECT 'E', 'INVALID PRICE_CHANGE_TRAN_TYPE: NOT IN ALLOWED VALUES', '20260610T201849_7ba6ad21', 'ERROR_INVALID_PRICE_CHANGE_TRAN_TYPE', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE PRICE_CHANGE_TRAN_TYPE NOT IN ('0', '2', '4', '8', '9', '10', '11')
UNION ALL
SELECT 'E', 'ITEM NOT FOUND: ITEM_MASTER_CTRL', '20260610T201849_7ba6ad21', 'ERROR_MISSING_ITEM', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE ITEM IS NOT NULL AND NOT EXISTS (SELECT 1 FROM ITEM_MASTER_CTRL d WHERE d.ITEM = s.ITEM AND CTRL_STATUS = 'C')
UNION ALL
SELECT 'E', 'ORG_NUM NOT FOUND: STORE_ADD_CTRL, WH_CTRL', '20260610T201849_7ba6ad21', 'ERROR_INVALID_ORG', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE ORG_NUM IS NOT NULL AND NOT EXISTS (SELECT 1 FROM DMF_VALIDITY__ORGANIZAT_A4FC2D d WHERE d.ORG_NUM = s.ORG_NUM)
UNION ALL
SELECT 'W', 'WARN_MISSING_ITEM_LOCATION', '20260610T201849_7ba6ad21', 'WARN_MISSING_ITEM_LOCATION', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE NOT EXISTS (SELECT 1 FROM ITEM_LOC_CTRL c WHERE c.ctrl_status = 'C' AND c.ITEM = s.ITEM AND c.LOC = s.ORG_NUM)
UNION ALL
SELECT 'W', 'WARN_ZERO_PRICE', '20260610T201849_7ba6ad21', 'WARN_ZERO_PRICE', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE SELLING_UNIT_RTL_AMT_LCL = 0
UNION ALL
SELECT 'E', 'DUPLICATE KEY: ITEM, ORG_NUM, DAY_DT', '20260610T201849_7ba6ad21', 'ERR_DUP_KEY_CONFLICT', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE FROM s WHERE (ITEM, ORG_NUM, DAY_DT) IN (SELECT ITEM, ORG_NUM, DAY_DT FROM DMF_DUPKEYS__PRICE_INIT_DELTA);
COMMIT;
-- ============================================================================
-- STEP 6 — Build typed INIT_PRICE_DELTA_STG (drop-and-recreate)
-- ============================================================================
-- DDL types are pulled from the RAP spec (not VARCHAR2(4000) blanket).
DROP TABLE INIT_PRICE_DELTA_STG PURGE; -- ignore ORA-00942
CREATE TABLE INIT_PRICE_DELTA_STG (
ITEM VARCHAR2(4000),
ORG_NUM VARCHAR2(4000),
DAY_DT VARCHAR2(4000),
PRICE_CHANGE_TRAN_TYPE VARCHAR2(4000),
STANDARD_UNIT_RTL_AMT_LCL VARCHAR2(4000),
SELLING_UNIT_RTL_AMT_LCL VARCHAR2(4000),
BASE_COST_AMT_LCL VARCHAR2(4000),
LOC_CURR_CODE VARCHAR2(4000),
DOC_CURR_CODE VARCHAR2(4000),
INSERTED_AT TIMESTAMP(6)
);
-- ============================================================================
-- STEP 7 — Populate INIT_PRICE_DELTA_STG (anti-join on SRC_ROWID; LOG_LVL='W' rows still flow through)
-- ============================================================================
INSERT INTO INIT_PRICE_DELTA_STG (ITEM, ORG_NUM, DAY_DT, PRICE_CHANGE_TRAN_TYPE, STANDARD_UNIT_RTL_AMT_LCL, SELLING_UNIT_RTL_AMT_LCL, BASE_COST_AMT_LCL, LOC_CURR_CODE, DOC_CURR_CODE, INSERTED_AT)
WITH s AS (
SELECT
SRC_ROWID,
ITEM,
ORG_NUM,
DAY_DT,
PRICE_CHANGE_TRAN_TYPE,
STANDARD_UNIT_RTL_AMT_LCL,
SELLING_UNIT_RTL_AMT_LCL,
BASE_COST_AMT_LCL,
LOC_CURR_CODE,
DOC_CURR_CODE
FROM DMF_SOURCE__PRICE_INIT_DELTA
)
SELECT s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, s.LOC_CURR_CODE, s.DOC_CURR_CODE, SYSDATE
FROM s
WHERE NOT EXISTS (
SELECT 1
FROM INIT_PRICE_DELTA_REJECTED r
WHERE r.VALIDATION_RUN_ID = '20260610T201849_7ba6ad21'
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
AND s.PRICE_CHANGE_TRAN_TYPE IS NOT NULL;
COMMIT;
-- Row count check: should be COUNT(INIT_PRICE_DELTA_RAW) minus distinct SRC_ROWIDs in INIT_PRICE_DELTA_REJECTED where LOG_LVL='E'.
SELECT COUNT(*) AS stg_rows FROM INIT_PRICE_DELTA_STG;
-- ============================================================================
-- STEP 8 — Build INIT_PRICE_DELTA_CTRL (drop-and-recreate)
-- ============================================================================
-- Final Oracle export shape. Column projection comes from
-- `[ctrl_projection]` in the override; default = identity from STG.
DROP TABLE INIT_PRICE_DELTA_CTRL PURGE; -- ignore ORA-00942
CREATE TABLE INIT_PRICE_DELTA_CTRL (
ITEM VARCHAR2(4000),
ORG_NUM VARCHAR2(4000),
DAY_DT DATE,
PRICE_CHANGE_TRAN_TYPE VARCHAR2(4000),
MULTI_SELLING_UOM VARCHAR2(4000),
SELLING_UOM VARCHAR2(4000),
MULTI_UNIT_QTY VARCHAR2(4000),
MULTI_UNIT_RTL_AMT_LCL VARCHAR2(4000),
STANDARD_UNIT_RTL_AMT_LCL VARCHAR2(4000),
SELLING_UNIT_RTL_AMT_LCL VARCHAR2(4000),
BASE_COST_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),
ORIG_SELLING_UNIT_RTL_AMT_LCL VARCHAR2(4000),
LST_REG_RTL_AMT_LCL VARCHAR2(4000),
INSERTED_AT TIMESTAMP(6)
);
-- ============================================================================
-- STEP 9 — Populate INIT_PRICE_DELTA_CTRL from INIT_PRICE_DELTA_STG
-- ============================================================================
INSERT INTO INIT_PRICE_DELTA_CTRL (ITEM, ORG_NUM, DAY_DT, PRICE_CHANGE_TRAN_TYPE, MULTI_SELLING_UOM, SELLING_UOM, MULTI_UNIT_QTY, MULTI_UNIT_RTL_AMT_LCL, STANDARD_UNIT_RTL_AMT_LCL, SELLING_UNIT_RTL_AMT_LCL, BASE_COST_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, ORIG_SELLING_UNIT_RTL_AMT_LCL, LST_REG_RTL_AMT_LCL, INSERTED_AT)
SELECT s.ITEM, s.ORG_NUM, s.DAY_DT, s.PRICE_CHANGE_TRAN_TYPE, CAST(NULL AS VARCHAR2(4)), CAST(NULL AS VARCHAR2(4)), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), s.STANDARD_UNIT_RTL_AMT_LCL, s.SELLING_UNIT_RTL_AMT_LCL, s.BASE_COST_AMT_LCL, CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), s.LOC_CURR_CODE, CAST(NULL AS NUMBER), s.DOC_CURR_CODE, CAST(NULL AS NUMBER), CAST(NULL AS CHAR(1)), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), SYSDATE
FROM INIT_PRICE_DELTA_STG s;
COMMIT;
SELECT COUNT(*) AS ctrl_rows FROM INIT_PRICE_DELTA_CTRL;
-- ============================================================================
-- STEP 10 — Rejection summary by rule
-- ============================================================================
SELECT RULE_ID, LOG_LVL, COUNT(*) AS rejected_rows
FROM INIT_PRICE_DELTA_REJECTED
GROUP BY RULE_ID, LOG_LVL
ORDER BY rejected_rows DESC;