Convert ODI mappings and packages to Snowflake SQL

<!-- ODI Mapping: Sync_Policy_Dim -->
<Mapping name="Sync_Policy_Dim">
  <Source schema="STAGING"
         table="STG_POLICIES"/>
  <Target schema="DW"
         table="DIM_POLICY"/>
  <Join condition="SRC.POLICY_ID = TGT.POLICY_ID"/>
  <Filter>SRC.EFF_DT >= :LAST_EXTRACT</Filter>
  <Expression target="POLICY_STATUS">
    DECODE(STATUS_CD,'A','Active','C','Cancelled',
           'L','Lapsed','Unknown')
  </Expression>
  <Expression target="ANNUAL_PREMIUM">
    NVL(MONTHLY_PREM,0) * 12
  </Expression>
  <IKM name="IKM Oracle Incremental Update"
       FLOW_CONTROL="true"/>
  <JKM name="JKM Oracle Consistent"
       SUBSCRIBER="DW_CDC"/>
</Mapping>

-- ODI IKM + JKM → Snowflake Stream + MERGE

-- Stream replaces JKM journalizing
CREATE STREAM IF NOT EXISTS stg_policies_stream
  ON TABLE staging.stg_policies;

-- Stored procedure replaces ODI mapping
CREATE OR REPLACE PROCEDURE dw.sync_policy_dim()
RETURNS STRING
LANGUAGE SQL
AS
$$
  MERGE INTO dw.dim_policy AS tgt
  USING (
    SELECT
      policy_id,
      CASE status_cd
        WHEN 'A' THEN 'Active'
        WHEN 'C' THEN 'Cancelled'
        WHEN 'L' THEN 'Lapsed'
        ELSE 'Unknown'
      END AS policy_status,
      COALESCE(monthly_prem, 0) * 12
        AS annual_premium
    FROM stg_policies_stream
    WHERE METADATA$ACTION = 'INSERT'
  ) AS src
  ON tgt.policy_id = src.policy_id
  WHEN MATCHED THEN UPDATE SET
    tgt.policy_status = src.policy_status,
    tgt.annual_premium = src.annual_premium
  WHEN NOT MATCHED THEN INSERT
    (policy_id, policy_status, annual_premium)
    VALUES
    (src.policy_id, src.policy_status,
     src.annual_premium);
  RETURN 'Sync complete';
$$;

-- Task replaces ODI package scheduling
CREATE OR REPLACE TASK dw.sync_policy_task
  WAREHOUSE = 'ETL_WH'
  SCHEDULE = 'USING CRON 0 */2 * * * UTC'
  WHEN SYSTEM$STREAM_HAS_DATA('stg_policies_stream')
AS CALL dw.sync_policy_dim();