dc/sas/sasjs/macros/mpe_targetloader.sas

/**
  @file
  @brief Generic loader for tables configured in &mpelib..mpe_tables
  @details General wrapper for loading tables using ALL load types.

  Called by admin/postdata.  The dclib should already be assigned.

  @param [in] dclib= (NOTPROVIDED) The data controller control library
  @param [in] libds= the library.dataset to load
  @param [in] now= (%sysfunc(datetime())) Static processed timestamp

  <h4> SAS Macros </h4>
  @li bitemporal_dataloader.sas
  @li mf_existvar.sas
  @li mf_getengine.sas
  @li mf_getuniquename.sas
  @li mf_getuser.sas
  @li mp_abort.sas
  @li mp_loadformat.sas
  @li mp_lockanytable.sas

  @version 9.2
  @author 4GL Apps Ltd
  @copyright 4GL Apps Ltd.  This code may only be used within Data Controller
    and may not be re-distributed or re-sold without the express permission of
    4GL Apps Ltd.

**/

%macro mpe_targetloader(libds= /* library.dataset to LOAD (target) */
    ,now= %sysfunc(datetime()) /* static processed timestamp */
    ,etlsource=                 /* process from whence the data came */
    ,STAGING_DS= STAGING_DS     /* name of staging (work) dataset which should
                                    be appended into the target. */
    ,LOADTARGET=NO              /* set to yes to actually load the target */
    ,CLOSE_VARS=                /* provide close vars to override defaults */
    ,dclib=NOTPROVIDED
    ,mdebug=0
    ,dc_dttmtfmt=%sysfunc(datetime())
  );
%local lib ds nobs;

/**
  * if a format catalog (suffix "-FC") we assume the catalog has already been
  *   created by the calling program with a libds of work.fmtextract
*/
%let orig_lib=%upcase(%scan(&libds,1,.));
%let orig_ds=%upcase(%scan(&libds,2,.));
%let orig_libds=&libds;
%if %scan(&libds,2,-)=FC %then %do;
  %let lib=WORK;
  %let ds=FMTEXTRACT;
  %let libds=&lib..&ds;
%end;
%else %do;
  %let lib=&orig_lib;
  %let ds=&orig_ds;
%end;

%mp_abort(iftrue= (&dclib=NOTPROVIDED)
  ,mac=&sysmacroname
  ,msg=%str(dclib=NOTPROVIDED)
)

/* get table attributes */
%let nobs=0;
data work.sumo_config;
  set &mpelib..mpe_tables;
  where &dc_dttmtfmt. lt tx_to
    and libref="&orig_lib"
    and dsn="&orig_ds";
  call symputx('LOADTYPE',loadtype,'l');
  call symputx('BUSKEY',buskey,'l');
  call symputx('VAR_TXFROM',var_txfrom,'l');
  call symputx('VAR_TXTO',var_txto,'l');
  call symputx('VAR_BUSFROM',var_busfrom,'l');
  call symputx('VAR_BUSTO',var_busto,'l');
  call symputx('VAR_PROCESSED',VAR_PROCESSED,'l');
  call symputx('RK_UNDERLYING',RK_UNDERLYING,'l');
  %if %length(&CLOSE_VARS)=0 %then %do;
    call symputx('CLOSE_VARS',CLOSE_VARS,'l');
  %end;
  call symputx('nobs',_n_,'l');
  if missing(AUDIT_LIBDS) then AUDIT_LIBDS="&dclib..MPE_AUDIT";
  call symputx('AUDIT_LIBDS',AUDIT_LIBDS,'l');
  put (_all_)(=);
run;

/* check if table is actually configured to load */
%if &nobs ne 1 %then %do;
  proc sql;
  insert into &mpelib..mpe_loads
    set USER_NM="%mf_getuser()"
      ,STATUS='FAILED (BAD DS)'
      ,CSV_DIR=symget('ETLSOURCE')
      ,PROCESSED_DTTM=&now;
%end;
%mp_abort(iftrue= (&nobs=0)
  ,mac=&sysmacroname
  ,msg=%str(Table not registered in &mpelib..mpe_tables)
)
%mp_abort(iftrue= (&nobs>1)
  ,mac=&sysmacroname
  ,msg=%str(Something is very wrong with the PK in &mpelib..mpe_tables)
)


%if &LOADTYPE=TXTEMPORAL %then %do;

  %bitemporal_dataloader(bus_from=,bus_to= /* explicitly empty*/
      ,tech_from=&VAR_TXFROM
      ,tech_to = &VAR_TXTO
      ,base_lib=&lib
      ,base_dsn=&ds
      ,append_lib=WORK
      ,append_dsn=&STAGING_DS
      ,high_date='31DEC9999:23:59:59'dt
      ,PK= &buskey
      ,ETLSOURCE=&ETLSOURCE
      ,LOADTYPE=&loadtype
      ,RK_UNDERLYING=&RK_UNDERLYING
      ,LOADTARGET=&LOADTARGET
      ,RK_UPDATE_MAXKEYTABLE=&LOADTARGET
      ,CLOSE_VARS=&CLOSE_VARS
      ,processed=&VAR_PROCESSED
      ,dclib=&dclib
      ,outds_audit=&AUDIT_LIBDS
  )
%end;
%else %if &loadtype=REPLACE %then %do;
  %if &LOADTARGET=YES %then %do;
    %mp_lockanytable(LOCK,lib=&lib,ds=&ds,ref=%str(&etlsource),
      ctl_ds=&dclib..mpe_lockanytable
    )
    data WORK.&STAGING_DS;
      set WORK.&STAGING_DS;
    %if %mf_existvar(&libds,&VAR_PROCESSED) %then %do;
      &VAR_PROCESSED = &now;
    %end;
      drop _____DELETE__THIS__RECORD_____;
    run;
    %local engine_type;
    %let engine_type=%mf_getengine(&lib);
    %if &engine_type=CAS %then %do;
      /* Fixed char variables cannot be appended to CAS varchar columns, so
        cast them (per the target table structure) in a CASUSER copy of the
        staging table, then append via data step.  Approach mirrors the CAS
        append in bitemporal_dataloader.sas */
      proc contents noprint data=&libds
        out=work.rpl_base_cols(keep=name type);
      run;
      proc contents noprint data=WORK.&STAGING_DS
        out=work.rpl_stag_cols(keep=name type);
      run;
      proc sql noprint;
      create table work.rpl_vchars as
        select a.name
        from work.rpl_base_cols a
        inner join work.rpl_stag_cols b
          on upcase(a.name)=upcase(b.name)
        where a.type=6; /* varchar in target */
      quit;
      /* get varchar variables ready for casting */
      %local vcfmt vcrename vcassign vcdrop tmpds;
      %let vcfmt=;
      %let vcrename=;
      %let vcassign=;
      %let vcdrop=;
      data _null_;
        set work.rpl_vchars end=last;
        length vcrename vcassign vcdrop vcfmt $32767 rancol $32;
        retain vcrename vcassign vcdrop vcfmt;
        if _n_=1 then vcrename='(rename=(';
        rancol=resolve('%mf_getuniquename()');
        vcfmt=trim(vcfmt)!!'length '!!cats(name)!!' varchar(*);';
        vcrename=trim(vcrename)!!' '!!cats(name,'=',rancol);
        vcassign=cats(vcassign,name,'=',rancol,';');
        vcdrop=cats(vcdrop,'drop '!!rancol,';');
        if last then do;
          vcrename=cats(vcrename,'))');
          call symputx('vcfmt',vcfmt);
          call symputx('vcrename',vcrename);
          call symputx('vcassign',vcassign);
          call symputx('vcdrop',vcdrop);
        end;
      run;
      /* prepare a temp cas table with varchars casted */
      %let tmpds=%mf_getuniquename();
      data casuser.&tmpds;
        &vcfmt
        set WORK.&STAGING_DS &vcrename;
        &vcassign
        &vcdrop
      run;
      /* exit on err condition before the destructive truncate below */
      %if &syscc>0 %then %do;
        %mp_lockanytable(UNLOCK,lib=&lib,ds=&ds,ref=&ETLSOURCE (aborted),
          ctl_ds=&dclib..mpe_lockanytable
        )
        /* ensure cleanup */
        proc sql;drop table CASUSER.&tmpds;quit;
      %end;
      %mp_abort(iftrue= (&syscc>0)
        ,mac=&sysmacroname
        ,msg=%str(syscc=&syscc - aborting before REPLACE truncate of &libds.)
      )
      /* CAS tables do not support SQL deletes, so truncate with deleteRows.
        This is deliberately the last step before the append, to minimise
        the time in which the target table is empty. */
      proc cas;
        table.deleteRows / table={caslib="&lib",name="&ds",where="1=1"};
      quit;
      /* load the target with varchars applied */
      data &libds (append=yes) / sessref=dcsession;
        set casuser.&tmpds;
      run;
      /* drop temp table */
      proc sql;
        drop table CASUSER.&tmpds;
      quit;
    %end;
    %else %do;
      /* exit on err condition before the destructive delete below */
      %if &syscc>0 %then %do;
        %mp_lockanytable(UNLOCK,lib=&lib,ds=&ds,ref=&ETLSOURCE (aborted),
          ctl_ds=&dclib..mpe_lockanytable
        )
      %end;
      %mp_abort(iftrue= (&syscc>0)
        ,mac=&sysmacroname
        ,msg=%str(syscc=&syscc - aborting before REPLACE delete of &libds.)
      )
      proc sql;
        delete * from &libds;
      quit;
      proc append base=&libds data=WORK.&STAGING_DS force nowarn;run;
    %end;

    %mp_lockanytable(UNLOCK,lib=&lib,ds=&ds,ctl_ds=&dclib..mpe_lockanytable)
  %end;
  %else %do;
    /* is full replace so treat all staged records as new in diff screen */
    data work.outds_mod work.outds_add ;
      set work.&staging_ds;
      output work.outds_add;
    run;
    /* previous table will be considered fully deleted */
    data work.outds_del;
      set &lib..&ds;
    run;
  %end;
%end;
%else %if &loadtype=UPDATE %then %do;
  %bitemporal_dataloader(bus_from=,bus_to=
      ,tech_from= ,tech_to = /* explicitly empty*/
      ,base_lib=&lib
      ,base_dsn=&ds
      ,append_lib=WORK
      ,append_dsn=&STAGING_DS
      ,high_date='31DEC9999:23:59:59'dt
      ,PK= &buskey
      ,ETLSOURCE=%superq(etlsource)
      ,LOADTYPE=UPDATE
      ,RK_UNDERLYING=&RK_UNDERLYING
      ,LOADTARGET=&LOADTARGET
      ,RK_UPDATE_MAXKEYTABLE=&LOADTARGET
      ,processed=&VAR_PROCESSED
      ,dclib=&dclib
      ,outds_audit=&AUDIT_LIBDS
  )
%end;
%else %if &loadtype=FORMAT_CAT %then %do;
  /**
    * run mp_formatload
    * inputs:
    *  - LOADTARGET
    *  - CATALOG
    *  - STAGEDATA
    *  - LOADAUDIT
    * outputs:
    *  work.outds_add
    *  work.outds_del
    *  work.outds_mod
    */
  %mp_loadformat(&orig_libds
    ,&staging_ds
    ,loadtarget=&LOADTARGET
    ,auditlibds=&AUDIT_LIBDS
    ,locklibds=&dclib..mpe_lockanytable
    ,delete_col=_____DELETE__THIS__RECORD_____
    ,outds_add=outds_add
    ,outds_del=outds_del
    ,outds_mod=outds_mod
    ,mdebug=&mdebug
  )
%end;
%else %if &loadtype=BITEMPORAL %then %do;
  %bitemporal_dataloader(bus_from=&VAR_BUSFROM,bus_to=&VAR_BUSTO
      ,tech_from=&VAR_TXFROM
      ,tech_to = &VAR_TXTO
      ,base_lib=&lib
      ,base_dsn=&ds
      ,append_lib=WORK
      ,append_dsn=&STAGING_DS
      ,high_date='31DEC9999:23:59:59'dt
      ,PK= &buskey
      ,ETLSOURCE=%superq(etlsource)
      ,LOADTYPE=BITEMPORAL
      ,RK_UNDERLYING=&RK_UNDERLYING
      ,LOADTARGET=&LOADTARGET
      ,RK_UPDATE_MAXKEYTABLE=&LOADTARGET
      ,CLOSE_VARS=&CLOSE_VARS
      ,processed=&VAR_PROCESSED
      ,dclib=&dclib
      ,outds_audit=&AUDIT_LIBDS
  )
%end;
%else %do;
  %put %str(WARN)ING: LOADTYPE &LOADTYPE not supported;
  %let syscc=4;
  %mp_abort(msg=LOADTYPE &LOADTYPE not supported,mac=mpe_targetloader.sas)
%end;

%mp_abort(iftrue= (&syscc ne 0)
  ,mac=&_program
  ,msg=%str(syscc=&syscc exiting MPE_TARGETLOADER macro)
)

%mend mpe_targetloader;