Source: DMF/PuC/AIF/FelipeBkp/markdown_history.sql
-- ============================================================================
-- MARKDOWN HISTORY — Full Validation Pipeline
-- ============================================================================
-- Fact: markdown_history
-- Source: ORI_MARKDOWN_RAW
-- BQ source: `puc-p-dataf-common`.export_oracle_migration.markdown_history
-- Rejected: ORI_MARKDOWN_HIST_REJECTED
-- STG (accepted): ORI_MARKDOWN_HIST_STG
-- Rules: 6
-- 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-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 'organization_csv' → DMF_VALIDITY__ORGANIZATION_CSV
DROP TABLE DMF_VALIDITY__ORGANIZATION_CSV PURGE; -- ignore ORA-00942 on first run
CREATE TABLE DMF_VALIDITY__ORGANIZATION_CSV AS
SELECT DISTINCT TO_CHAR(STORE) AS org_num
FROM STORE_ADD_CTRL
WHERE ctrl_status = 'C';
CREATE INDEX IX_DMF_VALIDITY__ORGANIZATION_ ON DMF_VALIDITY__ORGANIZATION_CSV (ORG_NUM);
-- Verify the validity tables populated:
SELECT 'DMF_VALIDITY__ORGANIZATION_CSV' AS dim, COUNT(*) FROM DMF_VALIDITY__ORGANIZATION_CSV;
-- ============================================================================
-- STEP 2 — Materialize normalized source into DMF_SOURCE__MARKDOWN_HISTORY
-- ============================================================================
-- One scan of RAW now → all subsequent rules read from a smaller, hotter,
-- indexed snapshot. Drops the per-rule N-scan cost on wide RAW tables.
DROP TABLE DMF_SOURCE__MARKDOWN_HISTORY PURGE; -- ignore ORA-00942 on first run
CREATE TABLE DMF_SOURCE__MARKDOWN_HISTORY PARALLEL 4 AS
SELECT /*+ PARALLEL(4) */
ROWID AS SRC_ROWID,
ITEM AS ITEM,
LTRIM(ORG_NUM, '0') AS ORG_NUM,
DAY_DT AS DAY_DT,
RTL_TYPE_CODE AS RTL_TYPE_CODE,
MKDN_AMT_LCL AS MKDN_AMT_LCL,
DOC_CURR_CODE AS DOC_CURR_CODE,
LOC_CURR_CODE AS LOC_CURR_CODE
FROM ORI_MARKDOWN_RAW;
-- Index on the row-identity tuple makes duplicate_key and the STG
-- anti-join cheap.
CREATE INDEX IX_DMF_SOURCE__MARKDOWN_HIST_RID ON DMF_SOURCE__MARKDOWN_HISTORY (ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE);
-- Reset parallel attribute so per-rule reads don't auto-parallelize.
ALTER TABLE DMF_SOURCE__MARKDOWN_HISTORY NOPARALLEL;
-- Sanity check:
SELECT COUNT(*) FROM DMF_SOURCE__MARKDOWN_HISTORY;
-- ============================================================================
-- STEP 3 — Materialize duplicate-key set into DMF_DUPKEYS__MARKDOWN_HISTORY
-- ============================================================================
-- 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__MARKDOWN_HISTORY PURGE; -- ignore ORA-00942 on first run
CREATE TABLE DMF_DUPKEYS__MARKDOWN_HISTORY AS
SELECT ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE FROM DMF_SOURCE__MARKDOWN_HISTORY
GROUP BY ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE
HAVING COUNT(*) > 1;
CREATE INDEX IX_DMF_DUPKEYS__MARKDOWN_HIS_KEY ON DMF_DUPKEYS__MARKDOWN_HISTORY (ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE);
-- Sanity check (size of the conflict set):
SELECT COUNT(*) FROM DMF_DUPKEYS__MARKDOWN_HISTORY;
-- ============================================================================
-- STEP 4 — Ensure ORI_MARKDOWN_HIST_REJECTED exists, then empty it
-- ============================================================================
BEGIN
EXECUTE IMMEDIATE 'CREATE TABLE ORI_MARKDOWN_HIST_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), RTL_TYPE_CODE VARCHAR2(4000), MKDN_AMT_LCL VARCHAR2(4000), DOC_CURR_CODE VARCHAR2(4000), LOC_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 ORI_MARKDOWN_HIST_REJECTED;
-- If TRUNCATE fails with ORA-00054 (orphan lock), use DELETE instead:
-- DELETE FROM ORI_MARKDOWN_HIST_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_MARKDOWN_HIST_REJECTED (
LOG_LVL, ERROR_MSG, VALIDATION_RUN_ID, RULE_ID, SRC_ROWID, ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE, MKDN_AMT_LCL, DOC_CURR_CODE, LOC_CURR_CODE, INSERTED_AT
)
WITH s AS (
SELECT
SRC_ROWID,
ITEM,
ORG_NUM,
DAY_DT,
RTL_TYPE_CODE,
MKDN_AMT_LCL,
DOC_CURR_CODE,
LOC_CURR_CODE
FROM DMF_SOURCE__MARKDOWN_HISTORY
)
SELECT 'E', 'ITEM IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_CURR_CODE, 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.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_CURR_CODE, 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.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_CURR_CODE, SYSDATE FROM s WHERE DAY_DT IS NULL
UNION ALL
SELECT 'E', 'RTL_TYPE_CODE IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_CURR_CODE, SYSDATE FROM s WHERE RTL_TYPE_CODE IS NULL
UNION ALL
SELECT 'E', 'INVALID RTL_TYPE_CODE: NOT IN ALLOWED VALUES', 'MANUAL_RUN', 'ERROR_INVALID_RTL_TYPE_CODE', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_CURR_CODE, SYSDATE FROM s WHERE RTL_TYPE_CODE NOT IN ('R', 'P', 'C')
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.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_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', 'MANUAL_RUN', 'ERROR_MISSING_STORE', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_CURR_CODE, SYSDATE FROM s WHERE ORG_NUM IS NOT NULL AND NOT EXISTS (SELECT 1 FROM DMF_VALIDITY__ORGANIZATION_CSV d WHERE d.ORG_NUM = s.ORG_NUM)
UNION ALL
SELECT 'W', 'WARN_NEGATIVE_MKDN_AMT', 'MANUAL_RUN', 'WARN_NEGATIVE_MKDN_AMT', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_CURR_CODE, SYSDATE FROM s WHERE MKDN_AMT_LCL IS NOT NULL AND MKDN_AMT_LCL < 0
UNION ALL
SELECT 'E', 'DUPLICATE KEY: ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE', 'MANUAL_RUN', 'ERR_DUP_KEY_CONFLICT', s.SRC_ROWID, s.ITEM, s.ORG_NUM, s.DAY_DT, s.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_CURR_CODE, SYSDATE FROM s WHERE (ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE) IN (SELECT ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE FROM DMF_DUPKEYS__MARKDOWN_HISTORY);
COMMIT;
-- ============================================================================
-- STEP 6 — Build typed ORI_MARKDOWN_HIST_STG (drop-and-recreate)
-- ============================================================================
-- DDL types are pulled from the RAP spec (not VARCHAR2(4000) blanket).
DROP TABLE ORI_MARKDOWN_HIST_STG PURGE; -- ignore ORA-00942
CREATE TABLE ORI_MARKDOWN_HIST_STG (
ITEM VARCHAR2(4000),
ORG_NUM VARCHAR2(4000),
DAY_DT VARCHAR2(4000),
RTL_TYPE_CODE VARCHAR2(4000),
MKDN_AMT_LCL VARCHAR2(4000),
DOC_CURR_CODE VARCHAR2(4000),
LOC_CURR_CODE VARCHAR2(4000),
INSERTED_AT TIMESTAMP(6)
);
-- ============================================================================
-- STEP 7 — Populate ORI_MARKDOWN_HIST_STG (anti-join on SRC_ROWID; LOG_LVL='W' rows still flow through)
-- ============================================================================
INSERT INTO ORI_MARKDOWN_HIST_STG (ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE, MKDN_AMT_LCL, DOC_CURR_CODE, LOC_CURR_CODE, INSERTED_AT)
WITH s AS (
SELECT
SRC_ROWID,
ITEM,
ORG_NUM,
DAY_DT,
RTL_TYPE_CODE,
MKDN_AMT_LCL,
DOC_CURR_CODE,
LOC_CURR_CODE
FROM DMF_SOURCE__MARKDOWN_HISTORY
)
SELECT s.ITEM, s.ORG_NUM, s.DAY_DT, s.RTL_TYPE_CODE, s.MKDN_AMT_LCL, s.DOC_CURR_CODE, s.LOC_CURR_CODE, SYSDATE
FROM s
WHERE NOT EXISTS (
SELECT 1
FROM ORI_MARKDOWN_HIST_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
AND s.RTL_TYPE_CODE IS NOT NULL;
COMMIT;
-- Row count check: should be COUNT(ORI_MARKDOWN_RAW) minus distinct SRC_ROWIDs in ORI_MARKDOWN_HIST_REJECTED where LOG_LVL='E'.
SELECT COUNT(*) AS stg_rows FROM ORI_MARKDOWN_HIST_STG;
-- ============================================================================
-- STEP 8 — Build ORI_MARKDOWN_HIST_CTRL (drop-and-recreate)
-- ============================================================================
-- Final Oracle export shape. Column projection comes from
-- `[ctrl_projection]` in the override; default = identity from STG.
DROP TABLE ORI_MARKDOWN_HIST_CTRL PURGE; -- ignore ORA-00942
CREATE TABLE ORI_MARKDOWN_HIST_CTRL (
ITEM VARCHAR2(4000),
ORG_NUM VARCHAR2(4000),
DAY_DT VARCHAR2(4000),
RTL_TYPE_CODE VARCHAR2(4000),
MKDN_QTY VARCHAR2(4000),
MKDN_AMT_LCL VARCHAR2(4000),
MKUP_QTY VARCHAR2(4000),
MKUP_AMT_LCL VARCHAR2(4000),
MKDN_CAN_QTY VARCHAR2(4000),
MKDN_CAN_AMT_LCL VARCHAR2(4000),
MKUP_CAN_QTY VARCHAR2(4000),
MKUP_CAN_AMT_LCL VARCHAR2(4000),
DOC_CURR_CODE VARCHAR2(4000),
LOC_CURR_CODE VARCHAR2(4000),
GLOBAL1_EXCHANGE_RATE VARCHAR2(4000),
GLOBAL2_EXCHANGE_RATE VARCHAR2(4000),
GLOBAL3_EXCHANGE_RATE VARCHAR2(4000),
LOC_EXCHANGE_RATE VARCHAR2(4000),
REF_NO_1 VARCHAR2(4000),
REF_NO_2 VARCHAR2(4000),
DELETE_FLG VARCHAR2(4000),
DATASOURCE_NUM_ID VARCHAR2(4000),
FLEX1_NUM_VALUE VARCHAR2(4000),
FLEX2_NUM_VALUE VARCHAR2(4000),
FLEX3_NUM_VALUE VARCHAR2(4000),
FLEX4_NUM_VALUE VARCHAR2(4000),
FLEX5_NUM_VALUE VARCHAR2(4000),
FLEX6_NUM_VALUE VARCHAR2(4000),
FLEX7_NUM_VALUE VARCHAR2(4000),
FLEX8_NUM_VALUE VARCHAR2(4000),
FLEX9_NUM_VALUE VARCHAR2(4000),
FLEX10_NUM_VALUE VARCHAR2(4000),
FLEX11_NUM_VALUE VARCHAR2(4000),
FLEX12_NUM_VALUE VARCHAR2(4000),
FLEX13_NUM_VALUE VARCHAR2(4000),
FLEX14_NUM_VALUE VARCHAR2(4000),
FLEX15_NUM_VALUE VARCHAR2(4000),
FLEX16_NUM_VALUE VARCHAR2(4000),
FLEX17_NUM_VALUE VARCHAR2(4000),
FLEX18_NUM_VALUE VARCHAR2(4000),
FLEX19_NUM_VALUE VARCHAR2(4000),
FLEX20_NUM_VALUE VARCHAR2(4000),
CREATE_ID VARCHAR2(4000),
CREATE_DATETIME VARCHAR2(4000),
INSERTED_AT TIMESTAMP(6)
);
-- ============================================================================
-- STEP 9 — Populate ORI_MARKDOWN_HIST_CTRL from ORI_MARKDOWN_HIST_STG
-- ============================================================================
INSERT INTO ORI_MARKDOWN_HIST_CTRL (ITEM, ORG_NUM, DAY_DT, RTL_TYPE_CODE, MKDN_QTY, MKDN_AMT_LCL, MKUP_QTY, MKUP_AMT_LCL, MKDN_CAN_QTY, MKDN_CAN_AMT_LCL, MKUP_CAN_QTY, MKUP_CAN_AMT_LCL, DOC_CURR_CODE, LOC_CURR_CODE, GLOBAL1_EXCHANGE_RATE, GLOBAL2_EXCHANGE_RATE, GLOBAL3_EXCHANGE_RATE, LOC_EXCHANGE_RATE, REF_NO_1, REF_NO_2, DELETE_FLG, DATASOURCE_NUM_ID, FLEX1_NUM_VALUE, FLEX2_NUM_VALUE, FLEX3_NUM_VALUE, FLEX4_NUM_VALUE, FLEX5_NUM_VALUE, FLEX6_NUM_VALUE, FLEX7_NUM_VALUE, FLEX8_NUM_VALUE, FLEX9_NUM_VALUE, FLEX10_NUM_VALUE, FLEX11_NUM_VALUE, FLEX12_NUM_VALUE, FLEX13_NUM_VALUE, FLEX14_NUM_VALUE, FLEX15_NUM_VALUE, FLEX16_NUM_VALUE, FLEX17_NUM_VALUE, FLEX18_NUM_VALUE, FLEX19_NUM_VALUE, FLEX20_NUM_VALUE, CREATE_ID, CREATE_DATETIME, INSERTED_AT)
SELECT s.ITEM, s.ORG_NUM, s.DAY_DT, s.RTL_TYPE_CODE, CAST(NULL AS NUMBER), s.MKDN_AMT_LCL, CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), s.DOC_CURR_CODE, s.LOC_CURR_CODE, CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS VARCHAR2(48)), CAST(NULL AS VARCHAR2(250)), CAST(NULL AS CHAR(1)), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), CAST(NULL AS NUMBER), 'DMF', SYSDATE, SYSDATE
FROM ORI_MARKDOWN_HIST_STG s;
COMMIT;
SELECT COUNT(*) AS ctrl_rows FROM ORI_MARKDOWN_HIST_CTRL;
-- ============================================================================
-- STEP 10 — Rejection summary by rule
-- ============================================================================
SELECT RULE_ID, LOG_LVL, COUNT(*) AS rejected_rows
FROM ORI_MARKDOWN_HIST_REJECTED
GROUP BY RULE_ID, LOG_LVL
ORDER BY rejected_rows DESC;