Source: DMF/PuC/AIF/FelipeBkp/store_traffic.sql
-- ============================================================================
-- STORE TRAFFIC — Full Validation Pipeline
-- ============================================================================
-- Fact: store_traffic
-- Source: ORI_STORE_TRAFFIC_RAW
-- BQ source: `puc-p-dataf-common`.export_oracle_migration.migration_store_traffic
-- Rejected: ORI_STORE_TRAFFIC_REJECTED
-- STG (accepted): ORI_STORE_TRAFFIC_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.
--
-- 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__STORE_TRAFFIC
-- ============================================================================
-- 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__STORE_TRAFFIC PURGE; -- ignore ORA-00942 on first run
CREATE TABLE DMF_SOURCE__STORE_TRAFFIC PARALLEL 4 AS
SELECT /*+ PARALLEL(4) */
ROWID AS SRC_ROWID,
LTRIM(ORG_NUM, '0') AS ORG_NUM,
DAY_DT AS DAY_DT,
CASE WHEN REGEXP_LIKE(HOUR_24_NUM, '^[0-9]+$') THEN TO_NUMBER(HOUR_24_NUM) END AS HOUR_24_NUM,
CASE WHEN REGEXP_LIKE(MINUTE_NUM, '^[0-9]+$') THEN TO_NUMBER(MINUTE_NUM) END AS MINUTE_NUM,
STORE_TRAFFIC AS STORE_TRAFFIC,
DATASOURCE_NUM_ID AS DATASOURCE_NUM_ID,
INTEGRATION_ID AS INTEGRATION_ID
FROM ORI_STORE_TRAFFIC_RAW;
-- Index on the row-identity tuple makes duplicate_key and the STG
-- anti-join cheap.
CREATE INDEX IX_DMF_SOURCE__STORE_TRAFFIC_RID ON DMF_SOURCE__STORE_TRAFFIC (ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM);
-- Reset parallel attribute so per-rule reads don't auto-parallelize.
ALTER TABLE DMF_SOURCE__STORE_TRAFFIC NOPARALLEL;
-- Sanity check:
SELECT COUNT(*) FROM DMF_SOURCE__STORE_TRAFFIC;
-- ============================================================================
-- STEP 3 — Materialize duplicate-key set into DMF_DUPKEYS__STORE_TRAFFIC
-- ============================================================================
-- 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__STORE_TRAFFIC PURGE; -- ignore ORA-00942 on first run
CREATE TABLE DMF_DUPKEYS__STORE_TRAFFIC AS
SELECT ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM FROM DMF_SOURCE__STORE_TRAFFIC
GROUP BY ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM
HAVING COUNT(*) > 1;
CREATE INDEX IX_DMF_DUPKEYS__STORE_TRAFFI_KEY ON DMF_DUPKEYS__STORE_TRAFFIC (ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM);
-- Sanity check (size of the conflict set):
SELECT COUNT(*) FROM DMF_DUPKEYS__STORE_TRAFFIC;
-- ============================================================================
-- STEP 4 — Ensure ORI_STORE_TRAFFIC_REJECTED exists, then empty it
-- ============================================================================
BEGIN
EXECUTE IMMEDIATE 'CREATE TABLE ORI_STORE_TRAFFIC_REJECTED ( LOG_LVL VARCHAR2(1), ERROR_MSG VARCHAR2(255), VALIDATION_RUN_ID VARCHAR2(80), RULE_ID VARCHAR2(120), SRC_ROWID VARCHAR2(64), ORG_NUM VARCHAR2(4000), DAY_DT VARCHAR2(4000), HOUR_24_NUM VARCHAR2(4000), MINUTE_NUM VARCHAR2(4000), STORE_TRAFFIC VARCHAR2(4000), DATASOURCE_NUM_ID VARCHAR2(4000), INTEGRATION_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_STORE_TRAFFIC_REJECTED;
-- If TRUNCATE fails with ORA-00054 (orphan lock), use DELETE instead:
-- DELETE FROM ORI_STORE_TRAFFIC_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_STORE_TRAFFIC_REJECTED (
LOG_LVL, ERROR_MSG, VALIDATION_RUN_ID, RULE_ID, SRC_ROWID, ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM, STORE_TRAFFIC, DATASOURCE_NUM_ID, INTEGRATION_ID, INSERTED_AT
)
WITH s AS (
SELECT
SRC_ROWID,
ORG_NUM,
DAY_DT,
HOUR_24_NUM,
MINUTE_NUM,
STORE_TRAFFIC,
DATASOURCE_NUM_ID,
INTEGRATION_ID
FROM DMF_SOURCE__STORE_TRAFFIC
)
SELECT 'E', 'INVALID FORMAT: ORG_NUM', 'MANUAL_RUN', 'ERROR_DTYPE_ORG_NUM', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_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.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE DAY_DT IS NOT NULL AND TRUNC(DAY_DT) <> DAY_DT
UNION ALL
SELECT 'E', 'INVALID FORMAT: HOUR_24_NUM', 'MANUAL_RUN', 'ERROR_DTYPE_HOUR_24_NUM', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE HOUR_24_NUM IS NOT NULL AND (ABS(TRUNC(TO_NUMBER(HOUR_24_NUM))) >= POWER(10,2) OR TO_NUMBER(HOUR_24_NUM) != ROUND(TO_NUMBER(HOUR_24_NUM),0))
UNION ALL
SELECT 'E', 'INVALID FORMAT: MINUTE_NUM', 'MANUAL_RUN', 'ERROR_DTYPE_MINUTE_NUM', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE MINUTE_NUM IS NOT NULL AND (ABS(TRUNC(TO_NUMBER(MINUTE_NUM))) >= POWER(10,2) OR TO_NUMBER(MINUTE_NUM) != ROUND(TO_NUMBER(MINUTE_NUM),0))
UNION ALL
SELECT 'E', 'INVALID FORMAT: STORE_TRAFFIC', 'MANUAL_RUN', 'ERROR_DTYPE_STORE_TRAFFIC', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE STORE_TRAFFIC IS NOT NULL AND (ABS(TRUNC(TO_NUMBER(STORE_TRAFFIC))) >= POWER(10,16) OR TO_NUMBER(STORE_TRAFFIC) != ROUND(TO_NUMBER(STORE_TRAFFIC),4))
UNION ALL
SELECT 'E', 'INVALID FORMAT: DATASOURCE_NUM_ID', 'MANUAL_RUN', 'ERROR_DTYPE_DATASOURCE_NUM_ID', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE DATASOURCE_NUM_ID IS NOT NULL AND (ABS(TRUNC(TO_NUMBER(DATASOURCE_NUM_ID))) >= POWER(10,10) OR TO_NUMBER(DATASOURCE_NUM_ID) != ROUND(TO_NUMBER(DATASOURCE_NUM_ID),0))
UNION ALL
SELECT 'E', 'INVALID FORMAT: INTEGRATION_ID', 'MANUAL_RUN', 'ERROR_DTYPE_INTEGRATION_ID', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE INTEGRATION_ID IS NOT NULL AND LENGTHB(TO_CHAR(INTEGRATION_ID)) > 80
UNION ALL
SELECT 'E', 'ORG_NUM IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_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.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE DAY_DT IS NULL
UNION ALL
SELECT 'E', 'HOUR_24_NUM IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE HOUR_24_NUM IS NULL
UNION ALL
SELECT 'E', 'MINUTE_NUM IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE MINUTE_NUM IS NULL
UNION ALL
SELECT 'E', 'DATASOURCE_NUM_ID IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE DATASOURCE_NUM_ID IS NULL
UNION ALL
SELECT 'E', 'INTEGRATION_ID IS NULL', 'MANUAL_RUN', 'ERROR_INVALID_NULL', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE INTEGRATION_ID IS NULL
UNION ALL
SELECT 'E', 'ERR_INVALID_HOUR_24_NUM', 'MANUAL_RUN', 'ERR_INVALID_HOUR_24_NUM', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE HOUR_24_NUM IS NOT NULL AND (HOUR_24_NUM < 0 OR HOUR_24_NUM > 23)
UNION ALL
SELECT 'E', 'ERR_INVALID_MINUTE_NUM', 'MANUAL_RUN', 'ERR_INVALID_MINUTE_NUM', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE MINUTE_NUM IS NOT NULL AND (MINUTE_NUM < 0 OR MINUTE_NUM > 59)
UNION ALL
SELECT 'E', 'ORG_NUM NOT FOUND: STORE_ADD_CTRL', 'MANUAL_RUN', 'ERROR_MISSING_STORE', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, 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_TRAFFIC', 'MANUAL_RUN', 'WARN_NEGATIVE_TRAFFIC', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE STORE_TRAFFIC IS NOT NULL AND STORE_TRAFFIC < 0
UNION ALL
SELECT 'E', 'DUPLICATE KEY: ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM', 'MANUAL_RUN', 'ERR_DUP_KEY_CONFLICT', s.SRC_ROWID, s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE FROM s WHERE (ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM) IN (SELECT ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM FROM DMF_DUPKEYS__STORE_TRAFFIC);
COMMIT;
-- ============================================================================
-- STEP 6 — Build typed ORI_STORE_TRAFFIC_STG (drop-and-recreate)
-- ============================================================================
-- DDL types are pulled from the RAP spec (not VARCHAR2(4000) blanket).
DROP TABLE ORI_STORE_TRAFFIC_STG PURGE; -- ignore ORA-00942
CREATE TABLE ORI_STORE_TRAFFIC_STG (
ORG_NUM VARCHAR2(80 BYTE),
DAY_DT DATE,
HOUR_24_NUM NUMBER(2,0),
MINUTE_NUM NUMBER(2,0),
STORE_TRAFFIC NUMBER(20,4),
DATASOURCE_NUM_ID NUMBER(10,0),
INTEGRATION_ID VARCHAR2(80 BYTE),
INSERTED_AT TIMESTAMP(6)
);
-- ============================================================================
-- STEP 7 — Populate ORI_STORE_TRAFFIC_STG (anti-join on SRC_ROWID; LOG_LVL='W' rows still flow through)
-- ============================================================================
INSERT INTO ORI_STORE_TRAFFIC_STG (ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM, STORE_TRAFFIC, DATASOURCE_NUM_ID, INTEGRATION_ID, INSERTED_AT)
WITH s AS (
SELECT
SRC_ROWID,
ORG_NUM,
DAY_DT,
HOUR_24_NUM,
MINUTE_NUM,
STORE_TRAFFIC,
DATASOURCE_NUM_ID,
INTEGRATION_ID
FROM DMF_SOURCE__STORE_TRAFFIC
)
SELECT s.ORG_NUM, s.DAY_DT, s.HOUR_24_NUM, s.MINUTE_NUM, s.STORE_TRAFFIC, s.DATASOURCE_NUM_ID, s.INTEGRATION_ID, SYSDATE
FROM s
WHERE NOT EXISTS (
SELECT 1
FROM ORI_STORE_TRAFFIC_REJECTED r
WHERE r.VALIDATION_RUN_ID = 'MANUAL_RUN'
AND r.LOG_LVL = 'E'
AND r.SRC_ROWID = s.SRC_ROWID
)
AND s.ORG_NUM IS NOT NULL
AND s.DAY_DT IS NOT NULL
AND s.HOUR_24_NUM IS NOT NULL
AND s.MINUTE_NUM IS NOT NULL
AND s.DATASOURCE_NUM_ID IS NOT NULL
AND s.INTEGRATION_ID IS NOT NULL;
COMMIT;
-- Row count check: should be COUNT(ORI_STORE_TRAFFIC_RAW) minus distinct SRC_ROWIDs in ORI_STORE_TRAFFIC_REJECTED where LOG_LVL='E'.
SELECT COUNT(*) AS stg_rows FROM ORI_STORE_TRAFFIC_STG;
-- ============================================================================
-- STEP 8 — Build ORI_STORE_TRAFFIC_CTRL (drop-and-recreate)
-- ============================================================================
-- Final Oracle export shape. Column projection comes from
-- `[ctrl_projection]` in the override; default = identity from STG.
DROP TABLE ORI_STORE_TRAFFIC_CTRL PURGE; -- ignore ORA-00942
CREATE TABLE ORI_STORE_TRAFFIC_CTRL (
ORG_NUM VARCHAR2(80 BYTE),
DAY_DT DATE,
HOUR_24_NUM NUMBER(2,0),
MINUTE_NUM NUMBER(2,0),
STORE_TRAFFIC NUMBER(20,4),
EXCHANGE_DT DATE,
AUX1_CHANGED_ON_DT DATE,
AUX2_CHANGED_ON_DT DATE,
AUX3_CHANGED_ON_DT DATE,
AUX4_CHANGED_ON_DT DATE,
CHANGED_BY_ID VARCHAR2(80 BYTE),
CHANGED_ON_DT DATE,
CREATED_BY_ID VARCHAR2(80 BYTE),
CREATED_ON_DT DATE,
DATASOURCE_NUM_ID NUMBER(10,0),
DELETE_FLG CHAR(1 BYTE),
DOC_CURR_CODE VARCHAR2(30 BYTE),
ETL_THREAD_VAL NUMBER(10,0),
GLOBAL1_EXCHANGE_RATE NUMBER(20,10),
GLOBAL2_EXCHANGE_RATE NUMBER(20,10),
GLOBAL3_EXCHANGE_RATE NUMBER(20,10),
INTEGRATION_ID VARCHAR2(80 BYTE),
LOC_CURR_CODE VARCHAR2(30 BYTE),
LOC_EXCHANGE_RATE NUMBER(20,10),
TENANT_ID VARCHAR2(80 BYTE),
X_CUSTOM VARCHAR2(4000 BYTE),
INSERTED_AT TIMESTAMP(6)
);
-- ============================================================================
-- STEP 9 — Populate ORI_STORE_TRAFFIC_CTRL from ORI_STORE_TRAFFIC_STG
-- ============================================================================
INSERT INTO ORI_STORE_TRAFFIC_CTRL (ORG_NUM, DAY_DT, HOUR_24_NUM, MINUTE_NUM, STORE_TRAFFIC, EXCHANGE_DT, AUX1_CHANGED_ON_DT, AUX2_CHANGED_ON_DT, AUX3_CHANGED_ON_DT, AUX4_CHANGED_ON_DT, CHANGED_BY_ID, CHANGED_ON_DT, CREATED_BY_ID, CREATED_ON_DT, DATASOURCE_NUM_ID, DELETE_FLG, DOC_CURR_CODE, ETL_THREAD_VAL, GLOBAL1_EXCHANGE_RATE, GLOBAL2_EXCHANGE_RATE, GLOBAL3_EXCHANGE_RATE, INTEGRATION_ID, LOC_CURR_CODE, LOC_EXCHANGE_RATE, TENANT_ID, X_CUSTOM, INSERTED_AT)
SELECT s.ORG_NUM, s.DAY_DT, CASE WHEN s.HOUR_24_NUM IS NULL OR ABS(TRUNC(s.HOUR_24_NUM)) < 100 THEN s.HOUR_24_NUM ELSE NULL END, CASE WHEN s.MINUTE_NUM IS NULL OR ABS(TRUNC(s.MINUTE_NUM)) < 100 THEN s.MINUTE_NUM ELSE NULL END, CASE WHEN s.STORE_TRAFFIC IS NULL OR ABS(TRUNC(s.STORE_TRAFFIC)) < 10000000000000000 THEN s.STORE_TRAFFIC ELSE NULL END, CAST(NULL AS DATE), CAST(NULL AS DATE), CAST(NULL AS DATE), CAST(NULL AS DATE), CAST(NULL AS DATE), CAST(NULL AS VARCHAR2(80 BYTE)), CAST(NULL AS DATE), CAST(NULL AS VARCHAR2(80 BYTE)), CAST(NULL AS DATE), CASE WHEN s.DATASOURCE_NUM_ID IS NULL OR ABS(TRUNC(s.DATASOURCE_NUM_ID)) < 10000000000 THEN s.DATASOURCE_NUM_ID ELSE NULL END, CAST(NULL AS CHAR(1 BYTE)), CAST(NULL AS VARCHAR2(30 BYTE)), CAST(NULL AS NUMBER(10,0)), CAST(NULL AS NUMBER(20,10)), CAST(NULL AS NUMBER(20,10)), CAST(NULL AS NUMBER(20,10)), s.INTEGRATION_ID, CAST(NULL AS VARCHAR2(30 BYTE)), CAST(NULL AS NUMBER(20,10)), CAST(NULL AS VARCHAR2(80 BYTE)), CAST(NULL AS VARCHAR2(4000 BYTE)), SYSDATE
FROM ORI_STORE_TRAFFIC_STG s;
COMMIT;
SELECT COUNT(*) AS ctrl_rows FROM ORI_STORE_TRAFFIC_CTRL;
-- ============================================================================
-- STEP 10 — Rejection summary by rule
-- ============================================================================
SELECT RULE_ID, LOG_LVL, COUNT(*) AS rejected_rows
FROM ORI_STORE_TRAFFIC_REJECTED
GROUP BY RULE_ID, LOG_LVL
ORDER BY rejected_rows DESC;