拉链表

create or replace procedure test.SP_SDI_S_CRM_CUST_PE_BASE_INFO (sp_today string,sp_job_id string,ret out int) is 
DECLARE  
        step INT;
        errmsg STRING;
BEGIN  
    BEGIN TRANSACTION
        ret := 0
        put_line(substr(current_date(), 1,10));
        step := 1;
        put_line(current_time() ||": STEP01: 重跑删除 test.SDI_S_CRM_CUST_PE_BASE_INFO" );
        DELETE FROM test.SDI_S_CRM_CUST_PE_BASE_INFO WHERE job_seq_id >= sp_job_id;
        put_line(current_time() ||": SETP01 重跑删除 test.SDI_S_CRM_CUST_PE_BASE_INFO 成功");
        step := step + 1 ;
        
        put_line(current_time() ||": STEP02: 重跑从历史表插入到 test.SDI_S_CRM_CUST_PE_BASE_INFO" );
        INSERT INTO test.SDI_S_CRM_CUST_PE_BASE_INFO SELECT cust_no,cert_type,cert_no,pecust_name,custname_short,cust_name_en,vip_cate,country,credit_level,is_freetax,btflbool,dimabool,sex,nty,mrg,btd,study_exp,degree,pay_acct,acct_wt_bk,year_income,fmy_ppl,inhbt_stat,cust_lev,bank_emp_ind,employee_typ,bank_stk_holder_ind,bank_party_ind,notes,xdzx_type,cust_qly,personal_insrc,insurance_dt,bad_record,is_ltc_credit_cust,native, eff_date,end_date,job_seq_id FROM test.HIS_S_CRM_CUST_PE_BASE_INFO WHERE new_job_seq_id=sp_job_id;  
        put_line(current_time() ||": SETP02 重跑从历史表插入到 test.SDI_S_CRM_CUST_PE_BASE_INFO 成功");
        step := step + 1 ;
        
        put_line(current_time() ||": STEP03: 重跑删除历史表原先数据 test.HIS_S_CRM_CUST_PE_BASE_INFO" );
        DELETE FROM test.HIS_S_CRM_CUST_PE_BASE_INFO WHERE new_job_seq_id>=sp_job_id;
        put_line(current_time() ||": SETP03 重跑删除历史表原先数据 test.HIS_S_CRM_CUST_PE_BASE_INFO 成功");
        step := step + 1 ;
        
        --保存状态到HIS (全字段比较不相等)
        put_line(current_time() ||": STEP04: 将数据备份到历史表 test.HIS_S_CRM_CUST_PE_BASE_INFO");
        INSERT INTO test.HIS_S_CRM_CUST_PE_BASE_INFO 
        SELECT cust_no,cert_type,cert_no,pecust_name,custname_short,cust_name_en,vip_cate,country,credit_level,is_freetax,btflbool,dimabool,sex,nty,mrg,btd,study_exp,degree,pay_acct,acct_wt_bk,year_income,fmy_ppl,inhbt_stat,cust_lev,bank_emp_ind,employee_typ,bank_stk_holder_ind,bank_party_ind,notes,xdzx_type,cust_qly,personal_insrc,insurance_dt,bad_record,is_ltc_credit_cust,native, eff_date,end_date,job_seq_id,sp_job_id 
        FROM test.SDI_S_CRM_CUST_PE_BASE_INFO N WHERE N.end_date='99990101' and not EXISTS (
        SELECT 1 
        FROM test.STG_S_CRM_CUST_PE_BASE_INFO M 
        where
         M.cust_no=N.cust_no AND M.cert_type=N.cert_type AND M.cert_no=N.cert_no AND M.pecust_name=N.pecust_name AND M.custname_short=N.custname_short AND M.cust_name_en=N.cust_name_en AND M.vip_cate=N.vip_cate AND M.country=N.country AND M.credit_level=N.credit_level AND M.is_freetax=N.is_freetax AND M.btflbool=N.btflbool AND M.dimabool=N.dimabool AND M.sex=N.sex AND M.nty=N.nty AND M.mrg=N.mrg AND M.btd=N.btd AND M.study_exp=N.study_exp AND M.degree=N.degree AND M.pay_acct=N.pay_acct AND M.acct_wt_bk=N.acct_wt_bk AND M.year_income=N.year_income AND M.fmy_ppl=N.fmy_ppl AND M.inhbt_stat=N.inhbt_stat AND M.cust_lev=N.cust_lev AND M.bank_emp_ind=N.bank_emp_ind AND M.employee_typ=N.employee_typ AND M.bank_stk_holder_ind=N.bank_stk_holder_ind AND M.bank_party_ind=N.bank_party_ind AND M.notes=N.notes AND M.xdzx_type=N.xdzx_type AND M.cust_qly=N.cust_qly AND M.personal_insrc=N.personal_insrc AND M.insurance_dt=N.insurance_dt AND M.bad_record=N.bad_record AND M.is_ltc_credit_cust=N.is_ltc_credit_cust AND M.native=N.native );
        put_line(current_time() ||": SETP04 将数据备份到历史表 test.HIS_S_CRM_CUST_PE_BASE_INFO 成功");
        step := step + 1 ;
        
        
        --新增数据直接insert
        put_line(current_time() ||": STEP05: 插入今日新增数据 test.SDI_S_CRM_CUST_PE_BASE_INFO" );
        INSERT INTO test.SDI_S_CRM_CUST_PE_BASE_INFO SELECT cust_no,cert_type,cert_no,pecust_name,custname_short,cust_name_en,vip_cate,country,credit_level,is_freetax,btflbool,dimabool,sex,nty,mrg,btd,study_exp,degree,pay_acct,acct_wt_bk,year_income,fmy_ppl,inhbt_stat,cust_lev,bank_emp_ind,employee_typ,bank_stk_holder_ind,bank_party_ind,notes,xdzx_type,cust_qly,personal_insrc,insurance_dt,bad_record,is_ltc_credit_cust,native, sp_today,'99990101',sp_job_id FROM test.STG_S_CRM_CUST_PE_BASE_INFO A where not EXISTS (select 1 from test.SDI_S_CRM_CUST_PE_BASE_INFO B where B.end_date='99990101' and A.CUST_NO=B.CUST_NO )
        put_line(current_time() ||": STEP05: 插入今日新增数据 test.SDI_S_CRM_CUST_PE_BASE_INFO 成功" );
        
        --当前批次做拉链
        put_line(current_time()||": STEP06: 当前批次做拉链 test.SDI_S_CRM_CUST_PE_BASE_INFO");
        UPDATE test.SDI_S_CRM_CUST_PE_BASE_INFO A SET END_DATE=sp_today,job_seq_id=sp_job_id WHERE END_DATE>=sp_today AND not EXISTS 
        (SELECT 1 FROM (SELECT M.CUST_NO FROM test.SDI_S_CRM_CUST_PE_BASE_INFO M
        inner join
        test.STG_S_CRM_CUST_PE_BASE_INFO N
        on M.cust_no=N.cust_no AND M.cert_type=N.cert_type AND M.cert_no=N.cert_no AND M.pecust_name=N.pecust_name AND M.custname_short=N.custname_short AND M.cust_name_en=N.cust_name_en AND M.vip_cate=N.vip_cate AND M.country=N.country AND M.credit_level=N.credit_level AND M.is_freetax=N.is_freetax AND M.btflbool=N.btflbool AND M.dimabool=N.dimabool AND M.sex=N.sex AND M.nty=N.nty AND M.mrg=N.mrg AND M.btd=N.btd AND M.study_exp=N.study_exp AND M.degree=N.degree AND M.pay_acct=N.pay_acct AND M.acct_wt_bk=N.acct_wt_bk AND M.year_income=N.year_income AND M.fmy_ppl=N.fmy_ppl AND M.inhbt_stat=N.inhbt_stat AND M.cust_lev=N.cust_lev AND M.bank_emp_ind=N.bank_emp_ind AND M.employee_typ=N.employee_typ AND M.bank_stk_holder_ind=N.bank_stk_holder_ind AND M.bank_party_ind=N.bank_party_ind AND M.notes=N.notes AND M.xdzx_type=N.xdzx_type AND M.cust_qly=N.cust_qly AND M.personal_insrc=N.personal_insrc AND M.insurance_dt=N.insurance_dt AND M.bad_record=N.bad_record AND M.is_ltc_credit_cust=N.is_ltc_credit_cust AND M.native=N.native  WHERE M.END_DATE='99990101') B 
        where A.CUST_NO=B.CUST_NO );
    
        put_line(current_time()||": STEP06 当前批次做拉链 test.SDI_S_CRM_CUST_PE_BASE_INFO 成功");
        step := step + 1;
        
        --当前批次拉链后新增
        put_line(current_time()||": STEP07: 当前批次拉链后新增 test.SDI_S_CRM_CUST_PE_BASE_INFO");
        INSERT INTO test.SDI_S_CRM_CUST_PE_BASE_INFO 
        SELECT cust_no,cert_type,cert_no,pecust_name,custname_short,cust_name_en,vip_cate,country,credit_level,is_freetax,btflbool,dimabool,sex,nty,mrg,btd,study_exp,degree,pay_acct,acct_wt_bk,year_income,fmy_ppl,inhbt_stat,cust_lev,bank_emp_ind,employee_typ,bank_stk_holder_ind,bank_party_ind,notes,xdzx_type,cust_qly,personal_insrc,insurance_dt,bad_record,is_ltc_credit_cust,native, sp_today,'99990101', sp_job_id from test.STG_S_CRM_CUST_PE_BASE_INFO A WHERE not EXISTS (SELECT 1 FROM  test.SDI_S_CRM_CUST_PE_BASE_INFO B
        where B.end_date = sp_today and A.CUST_NO=B.CUST_NO );
        
        put_line(current_time() ||": STEP07 当前批次拉链后新增 test.SDI_S_CRM_CUST_PE_BASE_INFO 成功");
        step := step + 1;
              
        COMMIT ;
        EXCEPTION 
            WHEN OTHERS THEN
                ret := -1
                put_line(sqlerrm());
                errmsg := CASE 
                WHEN step = 1 THEN ": STEP01: FAILD 重跑删除 test.SDI_S_CRM_CUST_PE_BASE_INFO"
                WHEN step = 2 THEN ": STEP02: FAILD 重跑从历史表插入到 test.SDI_S_CRM_CUST_PE_BASE_INFO"
                WHEN step = 3 THEN ": STEP03: FAILD 重跑删除历史表原先数据 test.HIS_S_CRM_CUST_PE_BASE_INFO"
                WHEN step = 4 THEN ": STEP04: FAILD 将数据备份到历史表 test.HIS_S_CRM_CUST_PE_BASE_INFO"
                WHEN step = 5 THEN ": STEP05: FAILD 插入今日新增数据 test.SDI_S_CRM_CUST_PE_BASE_INFO"
                WHEN step = 6 THEN ": STEP06: FAILD 当前批次做拉链 test.SDI_S_CRM_CUST_PE_BASE_INFO"
                WHEN step = 7 THEN ": STEP07: FAILD 当前批次拉链后新增 test.SDI_S_CRM_CUST_PE_BASE_INFO"
                END;
                put_line(current_time() || errmsg);

                ROLLBACK; 
END;