Loading...
Searching...
No Matches
bitemporal_dataloader.sas
Go to the documentation of this file.
1/**
2 @file
3 @brief Routine supporting multiple load types
4 @details Generic loader for multiple load types (UPDATE, SCD2, BITEMPORAL).
5
6 Handles all elements including metadata validation, PK checking, closeouts,
7 locking, logging, etc.
8
9 The staging table must be prepared with a unique business key. For bitemporal
10 this means a snapshot at both technical AND business time.
11
12ASSUMPTIONS:
13 - Base table has relevant datetime vars: 2xTechnical, 2xBusiness, 1xProcessed
14 - Staging table omits Technical or Processed datetimes (has Business only)
15 - Base table has no column names containing the string "___TMP___"
16 - Base &tech_from variable is not nullable. This should always be the case
17 anyway whenbuilding permanent bitemporal datasets.. But the point is that
18 this field is used to identify new records after the initial left join
19 from staging to base table.
20
21NOTES:
22 - All queries against BiTemporal tables need two filter conditions, as such:
23
24 where &bus_from LE [tstamp] LT &bus_to
25 AND &tx_from LE [tstamp] LT &tx_to
26
27 One cannot use BETWEEN
28 One cannot use &xx_from LE [tstamp] LE &xx_from (equivalent to above).
29 Background: https://stackoverflow.com/questions/20005950
30
31Areas for optimisation
32 - loading temporal history (currently experimental)
33
34 ## Supporting tables
35
36 Supporting tables must exist in the library specified in the `dclib` param.
37
38 ### MPE_DATALOADS
39
40 This table is updated every time a successful load occurs, and includes
41 information such as:
42
43 @li library
44 @li dataset
45 @li message (supplied in the ETLSOURCE param)
46 @li new rows
47 @li deleted rows
48 @li changed rows
49 @li timestamp
50 @li the user making the load
51 @li the version of (this) macro used to make the load
52
53
54 @param [in] APPEND_DSN= (APPENDTABLE) Name of STAGING table
55 @param [in] CONFIG_TABLE= (&dclib..MPE_CONFIG) The table containing library
56 engine specific config. The following scopes are supported:
57 @li DCBL_REDSH
58 @param [in] LOADTYPE= (BITEMPORAL) Supported types:
59 @li TXTEMPORAL - loads a buskey with version times
60 @li BUSTEMPORAL - loads buskey with bus + ver times
61 @li UPDATE - updates a buskey with NO history
62 @param [in] PROCESSED= (0) This column obtains a current timestamp for changed
63 records when loading the target table. Default is 0 (not set). If the
64 target table contains a variable called PROCESSED_DTTM, and processed=0,
65 then this column will be used for applying the current timestamp.
66 @param RK_MAXKEYTABLE= (mpe_maxkeyvalues) The maxkeytable to use (must exist
67 in DCLIB)
68 @param [in] PK= Business key, space separated. Should NOT include temporal
69 fields.
70 @param [in] RK_UNDERLYING= If supplied will generate an RK based on these
71 (space separated) business key fields. In this case only ONE PK field should
72 be supplied, which is assumed to be the RK. The RK field, plus underlying
73 fields, should all exist on the base table. The underlying fields should
74 exist on the staging table (the RK / PK field will be overwritten).
75 The staging table should also be unique on its PK.
76
77 @param [in] dclib= (&dc_libref) The library containing DC configuration tables
78 @param [out] outds_del= (work.outds_del) Output table containing
79 deleted records
80 @param [out] outds_add= (work.outds_add) Output table containing
81 appended records
82 @param [out] outds_mod= (work.outds_mod) Output table containing
83 changed records
84 @param [out] outds_audit= (0) Load detailed changes to an audit table. Uses
85 the mp_storediffs.sas macro. Provide the base table here, to load.
86
87 <h4> Global Variables </h4>
88 The following global macro variables are used. These should be replaced by
89 macro parameters in future releases.
90
91 @li `dc_dttmtfmt`
92
93 <h4> SAS Macros </h4>
94 @li bitemporal_closeouts.sas
95 @li dc_assignlib.sas
96 @li mf_existds.sas
97 @li mf_existvar.sas
98 @li mf_fmtdttm.sas
99 @li mf_getattrn.sas
100 @li mf_getengine.sas
101 @li mf_getschema.sas
102 @li mf_getuniquefileref.sas
103 @li mf_getuniquename.sas
104 @li mf_getuser.sas
105 @li mf_getvarlist.sas
106 @li mf_verifymacvars.sas
107 @li mf_wordsinstr1butnotstr2.sas
108 @li mp_abort.sas
109 @li mp_dropmembers.sas
110 @li mp_lockanytable.sas
111 @li mp_lockfilecheck.sas
112 @li mp_retainedkey.sas
113 @li mp_storediffs.sas
114 @li mp_rowhash.sas
115
116 @version 9.3
117 @author 4GL Apps Ltd.
118 @copyright 4GL Apps Ltd. This code may only be used within Data Controller
119 and may not be re-distributed or re-sold without the express permission of
120 4GL Apps Ltd.
121
122 @warning multitemporal loads (bitemporal for multiple points in business time)
123 are in experimental stage
124
125**/
126
127%macro bitemporal_dataloader(
128 bus_from= /* Business FROM datetime variable. Req'd on
129 STAGING & BASE tables.*/
130 ,bus_to = /* Business TO datetime variable. Req'd on
131 STAGING & BASE tables. */
132 ,bus_from_override= /* Provide a hard coded BUS_FROM datetime value.*/
133 ,bus_to_override= /* provide a hard coded BUS_TO datetime value */
134 ,tech_from= /* Technical FROM datetime variable. Req'd on
135 BASE table only. */
136 ,tech_to = /* Technical TO datetime variable. Req'd on BASE
137 table only. */
138 ,processed= 0
139 ,base_lib=WORK /* Libref of the BASE table. */
140 ,base_dsn=BASETABLE /* Name of BASE table. */
141 ,append_lib=WORK /* Libref of the STAGING table. */
142 ,append_dsn=APPENDTABLE
143 ,high_date='01JAN5999:00:00:00'dt /* High date to close out records */
144 ,PK= name sex
145 ,RK_UNDERLYING=
146 ,KEEPVARS= /* Provides option for removing unwanted vars from append table */
147 ,RK_UPDATE_MAXKEYTABLE=NO /* If switching (or mix matching) with regular
148 SCD2 loader then set this switch to YES to
149 ensure the MAXKEYTABLE is updated with the
150 current maximum RK value for the target table
151 */
152 ,CHECK_UNIQUENESS=YES /* Perform a check of the APPEND table to ensure it is
153 unique on its business key */
154 ,ETLSOURCE=demo /* supply a value ($50.) to show as ETLSOURCE in
155 &dclib..DATALOADS */
156 ,LOADTYPE=BITEMPORAL
157 ,RK_MAXKEYTABLE= mpe_maxkeyvalues
158 ,LOG=1 /* Switch to 0 to prevent records being added to
159 &mpelib..mpe_DATALOADS (ie when testing)*/
160 ,DELETE_COL= _____DELETE__THIS__RECORD_____
161 /* If this variable is found in the append dataset
162 then records are closed out (or deleted) in the
163 append table where that variable= "Yes" */
164 ,LOADTARGET=YES /* set to anything but uppercase YES to switch off
165 target table load and generate temp tables only */
166 ,CLOSE_VARS=
167/*a problem with regular SCD2 or TXTEMPORAL loads is that there is
168 no facility to close out removed records (all records are
169 assumed new or changed). But how does one determine which
170 records are removed? Short of loading the entire table
171 each time? This parameter allows a set of variables
172 (this should be a subset of the PK) to be declared, and
173 the macro will determine which records in the base table
174 need to be closed out ahead of the load.
175
176 For instance, given the following:
177
178 Base Table Staging Table
179 DATE ENTITY AMOUNT DATE ENTITY AMOUNT
180 JAN ACME4 66 JAN ACME4 66
181 FEB ACME4 99 FEB ACME4 99
182 FEB ACME1 22
183
184 By supplying DATE in CLOSE_VARS and DATE ENTITY as the PK,
185 the "FEB PAG 22" record would get closed out.
186 */
187 ,config_table=&dclib..MPE_CONFIG
188 ,dclib=&dc_libref
189 ,outds_del=work.outds_del
190 ,outds_add=work.outds_add
191 ,outds_mod=work.outds_mod
192 ,outds_audit=0
193 );
194
195/* when changing this macro, update the version num here */
196%local ver;
197%let ver=33;
198%put &sysmacroname entry vars:;
199%put _local_;
200
201%dc_assignlib(WRITE,&base_lib) /* may not already be assigned */
202
203/* return straight away if nothing to load */
204%let nobs= %mf_getattrn(&append_lib..&append_dsn,NLOBS);
205%if &nobs=-1 %then %do;
206 proc sql noprint; select count(*) into: nobs from &append_lib..&append_dsn;
207%end;
208%if &nobs=0 %then %do;
209 %put NOTE:; %put NOTE-;%put NOTE-;%put NOTE-;
210 %put NOTE- Base dataset &append_lib..&append_dsn is empty. Nothing to upload!;
211 %put NOTE-;%put NOTE-;%put NOTE-;
212 %return;
213%end;
214
215/* hard exit if err condition exists */
216%mp_abort(iftrue= (&syscc > 0)
217 ,mac=bitemporal_dataloader
218 ,msg=%str(Bitemporal transform / job aborted due to SYSCC=&SYSCC status;)
219)
220
221%local engine_type;
222%let engine_type=%mf_getengine(&base_lib);
223%if %length(&CLOSE_VARS)>0 and (&engine_type=REDSHIFT or &engine_type=POSTGRES
224or &engine_type=SNOW or &engine_type=SASIOSNF)
225%then %do;
226 %put NOTE:; %put NOTE-;%put NOTE-;%put NOTE-;
227 %put NOTE- CLOSE_VARS functionality not yet supported in &engine_type;
228 %put NOTE-;%put NOTE-;%put NOTE-;
229 %return;
230%end;
231
232/**
233 * The metadata functions (eg mf_existvar) will fail if the base table has a
234 * SAS lock. So, make a snapshot of the base table for further use.
235 * Also, make output tables (regardless).
236 */
237%local basecopy;
238%let basecopy=%mf_getuniquename(prefix=basecopy);
239
240data &basecopy &outds_mod &outds_add &outds_del;
241 set &base_lib..&base_dsn;
242 stop;
243run;
244%mp_abort(iftrue= (&syscc > 0)
245 ,mac=&_program
246 ,msg=%str(syscc=&syscc after base table copy - aborting due to table lock)
247)
248
249
250%local cols idx_pk md5_col hash_char_vars hash_num_vars ;
251%let md5_col=___TMP___md5;
252%let check_uniqueness=%upcase(&check_uniqueness);
253%let RK_UPDATE_MAXKEYTABLE=%upcase(&RK_UPDATE_MAXKEYTABLE);
254%let high_date=%unquote(&high_date);
255%let loadtype=%upcase(&loadtype);
256
257/* ensure irrelevant variables are cleared */
258%if &loadtype=BUSTEMPORAL %then %do;
259 %let tech_from=;
260 %let tech_to=;
261%end;
262%else %if &loadtype=TXTEMPORAL or &loadtype=UPDATE %then %do;
263 %let bus_from=;
264 %let bus_to=;
265%end;
266
267/* ensure relevant variables are supplied */
268%mp_abort(iftrue=(&loadtype=BITEMPORAL & %mf_verifymacvars(bus_from bus_to)=0)
269 ,mac=bitemporal_dataloader
270 ,msg=%str(Missing BUS_FROM / BUS_TO)
271)
272%mp_abort(iftrue=(&loadtype=TXTEMPORAL & %mf_verifymacvars(tech_from tech_to)=0)
273 ,mac=bitemporal_dataloader
274 ,msg=%str(Missing TECH_FROM / TECH_TO)
275)
276
277/**
278 * drop any tables (may be defined as views or vice versa preventing overwrite)
279 */
280%mp_dropmembers(append bitemp0_append bitemp_cols)
281
282/* SQL Server requires its own time values */
283/* 9.2 will only give picture format down to seconds. 9.3 allows
284 milliseconds by using lower S and defining the decimal in the format name..*/
285PROC FORMAT;
286 picture MyMSdt other='%0Y-%0m-%0dT%0H:%0M:%0S' (datatype=datetime);
287RUN;
288%local dbnow;
289%let dbnow="%sysfunc(datetime(),%mf_fmtdttm())"dt;
290
291data _null_;
292 /* convert space separated macvar to comma separated for SQL processing */
293 call symputx('PK_COMMA',tranwrd(compbl("&pk"),' ',','),'L');
294 call symputx('PK_CNT',countw("&pk",' '),'L');
295 now=&dbnow;
296 call symputx('NOW',now,'L');
297 call symputx('SQLNOW',cats("'",put(now,MyMSdt.),"'"),'L');
298 length etlsource $100;
299 etlsource=subpad(symget('etlsource'),1,100);
300 call symputx('etlsource',etlsource,'l');
301run;
302
303/**
304 * Even if no PROCESSED var provided, assume that any variable named
305 * PROCESSED_DTTM should be updated
306 */
307%if &processed=0 %then %do;
308 %if %mf_existvar(&basecopy,PROCESSED_DTTM)
309 %then %let processed=PROCESSED_DTTM;
310 %else %let processed=;
311%end;
312
313
314/* extract colnames for md5 creation / change tracking */
315proc contents noprint data=&base_lib..&base_dsn
316 out=work.bitemp_cols (keep=name type length varnum format:);
317run;
318proc sql noprint;
319select name into: cols separated by ','
320 from work.bitemp_cols
321 where upcase(name) not in
322 (%upcase("&bus_from","&bus_to"
323 ,"&tech_from","&tech_to"
324 ,"&processed","&delete_col")) ;
325/* Character variables are hashed separately from numerics so that the
326 iterative hash can be built with arrays rather than by concatenating
327 hundreds of 32-byte hex strings. */
328select name into: hash_char_vars separated by ' '
329 from work.bitemp_cols
330 where type in (2,6)
331 and upcase(name) not in
332 (%upcase("&bus_from","&bus_to"
333 ,"&tech_from","&tech_to"
334 ,"&processed","&delete_col")) ;
335select name into: hash_num_vars separated by ' '
336 from work.bitemp_cols
337 where type not in (2,6)
338 and upcase(name) not in
339 (%upcase("&bus_from","&bus_to"
340 ,"&tech_from","&tech_to"
341 ,"&processed","&delete_col")) ;
342
343/* set default formats*/
344%let bus_from_fmt = datetime19.;
345%let bus_to_fmt = datetime19.;
346%let processed_fmt = datetime19.;
347
348%let tech_from_fmt = format=datetime19.;
349%let tech_to_fmt = format=datetime19.;
350
351
352%put &=hash_char_vars;
353%put &=hash_num_vars;
354%put &=pk;
355
356data _null_;
357 set work.bitemp_cols;
358 if type=2 or type=6 then do;
359 length fmt $49.;
360 if format='' then fmt=cats('$',length,'.');
361 else fmt=cats(format,formatl,'.');
362 end;
363 else do;
364 if format='' then fmt=cats(length,'.');
365 else fmt=cats(format,formatl,'.',formatd);
366 end;
367 if upcase(name)="%upcase(&bus_from)" then
368 call symputx('bus_from_fmt',fmt,'L');
369 else if upcase(name)="%upcase(&bus_to)" then
370 call symputx('bus_to_fmt',fmt,'L');
371 else if upcase(name)="%upcase(&tech_from)" then
372 call symputx('tech_from_fmt',"format="!!fmt,'L');
373 else if upcase(name)="%upcase(&tech_to)" then
374 call symputx('tech_to_fmt',"format="!!fmt,'L');
375 else if upcase(name)="%upcase(&processed)" then
376 call symputx('processed_fmt',fmt,'L');
377run;
378
379%if %index(%quote(&cols),___TMP___) or %index(%quote(&cols),_____) %then %do;
380 %let msg=%str(Table contains a variable name containing "___TMP___" or "_____".%trim(
381 ) This may conflict with temp variable generation!!);
382 %mp_abort(msg=&msg,mac=bitemporal_dataloader);
383 %let syscc=5;
384 %return;
385%end;
386
387/* if transaction dates appear on the APPEND table, need to remove them */
388%local drop_tx_dates /* used in append table */
389 drop_tx_dates_noobs /* used to take the base table structure */;
390%if %mf_existvar(&append_lib..&append_dsn, &tech_from)
391 %then %let drop_tx_dates=&tech_from;
392%if %mf_existvar(&append_lib..&append_dsn, &tech_to)
393 %then %let drop_tx_dates=&drop_tx_dates &tech_to;
394%if %length(%trim(&drop_tx_dates))>0
395 %then %let drop_tx_dates=(drop=&drop_tx_dates);
396
397%if %mf_existvar(&basecopy, &tech_from)
398 %then %let drop_tx_dates_noobs=&tech_from;
399%if %mf_existvar(&basecopy, &tech_to)
400 %then %let drop_tx_dates_noobs=&drop_tx_dates_noobs &tech_to;
401%if %length(%trim(&drop_tx_dates_noobs))>0
402 %then %let drop_tx_dates_noobs=(drop=&drop_tx_dates_noobs obs=0);
403%else %let drop_tx_dates_noobs=(obs=0);
404
405
406/**
407 * Lock the table. This is necessary as we are doing a two part update (first
408 * closing records then appending new records). It is theoretically possible
409 * that an upload may occur whilst preparing the staging tables. And the
410 * staging tables are about to be prepared..
411 */
412%if &LOADTARGET = YES %then %do;
413 %put locking &base_lib..&base_dsn;
414 %mp_lockanytable(LOCK,
415 lib=&base_lib,ds=&base_dsn,ref=&ETLSOURCE,ctl_ds=&dclib..mpe_lockanytable
416 )
417 %if "&outds_audit" ne "0" %then %do;
418 %put locking &outds_audit;
419 %mp_lockanytable(LOCK
420 ,lib=%scan(&outds_audit,1,.)
421 ,ds=%scan(&outds_audit,2,.)
422 ,ref=&ETLSOURCE
423 ,ctl_ds=&dclib..mpe_lockanytable
424 )
425 %end;
426%end;
427%else %do;
428 /* not an actual load, so avoid updating the max key table in next step. */
429 %let rk_update_maxkeytable=NO;
430%end;
431
432%if %length(&RK_UNDERLYING)>0 %then %do;
433 %mp_retainedkey(
434 base_lib=&base_lib
435 ,base_dsn=&base_dsn
436 ,append_lib=&append_lib
437 ,append_dsn=&append_dsn
438 ,retained_key=&pk
439 ,business_key=&rk_underlying
440 ,check_uniqueness=&CHECK_UNIQUENESS
441 ,outds=work.append
442 %if &rk_update_maxkeytable=NO %then %do;
443 ,maxkeytable=0
444 %end;
445 %else %do;
446 ,maxkeytable=&dclib..&RK_MAXKEYTABLE
447 %end;
448 ,locktable=&dclib..mpe_lockanytable
449 %if &loadtype=BITEMPORAL or &loadtype=TXTEMPORAL %then %do;
450 ,filter_str=%str( (where=( &now < &tech_to)) )
451 %end;
452 )
453%end;
454%else %do;
455 proc sql;
456 create view work.append as select * from &append_lib..&append_dsn;
457%end;
458/**
459* generate md5 for append table
460*/
461/* it is possible the source dataset has additional (unwanted) columns.
462 Drop if specified; */
463%if %length(&keepvars)>0 %then %do;
464 /* remove tech dates from keepvars as they are generated later */
465 %let keepvars=%sysfunc(tranwrd(%str( &keepvars ),%str( &tech_from ),%str( )));
466 %let keepvars=%sysfunc(tranwrd(%str( &keepvars ),%str( &tech_to ),%str( )));
467 %let keepvars=(keep=&keepvars &bus_from &bus_to &processed &md5_col);
468%end;
469
470/* CAS varchar types cause append issues here, so perform autoconvert
471 by creating empty local table first */
472data;
473 set &base_lib..&base_dsn &drop_tx_dates_noobs;
474run;
475%local emptybasetable; %let emptybasetable=&syslast;
476
477data work.bitemp0_append &keepvars &outds_del(drop=&md5_col )
478 %if "%substr(&sysver,1,1)" ne "4" and "%substr(&sysver,1,1)" ne "5" %then %do;
479 /nonote2err
480 %end;
481 ;
482 /* apply formats for bitemporal vars but not tx dates which are added later */
483 %if %length(&keepvars)>0 and &loadtype=BITEMPORAL %then %do;
484 format &bus_from &bus_from_fmt;
485 format &bus_to &bus_to_fmt;
486 %end;
487 set &emptybasetable /* base table reqd in case append has fewer cols */
488 work.append &drop_tx_dates;
489 %if %length(%str(&bus_from_override))>0 %then %do;
490 &bus_from= %unquote(&bus_from_override) ;
491 %end;
492 %if %length(%str(&bus_to_override))>0 %then %do;
493 &bus_to= %unquote(&bus_to_override) ;
494 %end;
495 length &md5_col $32;
496 %mp_rowhash(md5_col=&md5_col
497 ,cvars=&hash_char_vars
498 ,nvars=&hash_num_vars
499 )
500 %if %length(&processed)>0 %then %do;
501 format &processed &processed_fmt;
502 &processed=&now;
503 %end;
504
505/**
506 * If a delete column exists then create the delete dataset
507 */
508%if %mf_existvar(&append_lib..&append_dsn, &delete_col) %then %do;
509 drop &delete_col;
510 if upcase(&delete_col) = "YES" then output &outds_del ;
511 else output work.bitemp0_append ;
512 run;
513
514 %if %mf_getattrn(&outds_del,NLOBS)>0 %then %do;
515 %bitemporal_closeouts(
516 tech_from=&tech_from
517 ,tech_to = &tech_to
518 ,base_lib=&base_lib
519 ,base_dsn=&base_dsn
520 ,append_lib=work
521 ,append_dsn=%scan(&outds_del,-1,.)
522 ,PK=&bus_from &pk
523 ,NOW=&dbnow
524 ,loadtarget=&loadtarget
525 ,loadtype=&loadtype
526 ,AUDITFOLDER=&dc_staging_area/&ETLSOURCE
527 )
528 %end;
529%end;
530%else %do;
531 output work.bitemp0_append;
532 run;
533%end;
534
535%mp_abort(iftrue= (&syscc gt 0 at line 494)
536 ,mac=&_program
537 ,msg=%str(syscc=&syscc)
538)
539
540%if %length(&close_vars)>0 %then %do;
541 /**
542 * need to close out records that are not provided
543 */
544 proc sql;
545 create table bitemp1_closevars1 as
546 select distinct a.%mf_getquotedstr(in_str=&pk,dlm=%str(,a.),quote=)
547 from &base_lib..&base_dsn a
548 inner join work.bitemp0_append b
549 on 1=1
550 /* join on closevars key */
551 %do idx_pk=1 %to %sysfunc(countw(&close_vars));
552 %let idx_val=%scan(&close_vars,&idx_pk);
553 and a.&idx_val=b.&idx_val
554 %end;
555 /* filter base on tech dates if necessary */
556 %if &loadtype=TXTEMPORAL %then %do;
557 where a.&tech_from <=&now and &now < a.&tech_to
558 %end;
559 ;
560 create table bitemp1_closevars2 as
561 select distinct a.*
562 from bitemp1_closevars1 a
563 left join work.bitemp0_append b
564 on 1=1
565 /* join on primary key */
566 %do idx_pk=1 %to %sysfunc(countw(&pk));
567 %let idx_val=%scan(&pk,&idx_pk);
568 and a.&idx_val=b.&idx_val
569 %end;
570 /* identify removed records by null value in a field in PK but not close_vars
571 */
572 where b.%scan(
573 %mf_wordsInStr1ButNotStr2(Str1=&pk,Str2=&close_vars),1,%str( )
574 ) IS NULL
575 ;
576
577 %if %mf_getattrn(bitemp1_closevars2,NLOBS)>0 %then %do;
578 %bitemporal_closeouts(
579 tech_from=&tech_from
580 ,tech_to = &tech_to
581 ,base_lib=&base_lib
582 ,base_dsn=&base_dsn
583 ,append_lib=work
584 ,append_dsn=bitemp1_closevars2
585 ,PK=&bus_from &pk
586 ,NOW=&dbnow
587 ,loadtarget=&loadtarget
588 ,loadtype=&loadtype
589 ,AUDITFOLDER=&dc_staging_area/&ETLSOURCE
590 )
591 %end;
592%end;
593
594/* return if nothing to load (was just deletes) */
595%if %mf_getattrn(work.bitemp0_append,NLOBS)=0 %then %do;
596 %put NOTE:; %put NOTE-;%put NOTE-;%put NOTE-;
597 %put NOTE- No updates - just deletes!;
598 %put NOTE-;%put NOTE-;%put NOTE-;
599%end;
600
601
602/**
603 * If applying manual overrides to business dates, then the input table MUST
604 * be unique on the PK. Check, and if not - abort.
605 */
606%local msg;
607%if %length(&bus_from_override.&bus_to_override)>0 or &CHECK_UNIQUENESS=YES
608%then %do;
609 proc sort data=work.bitemp0_append out=work.bitemp0_check nodupkey;
610 by &pk;
611 run;
612 %if %mf_getattrn(work.bitemp0_check,NLOBS)
613 ne %mf_getattrn(work.bitemp0_append,NLOBS)
614 %then %do;
615 %let msg=INPUT table &append_lib..&append_dsn is not unique on PK (&pk);
616 %mp_lockanytable(UNLOCK,lib=&base_lib,ds=&base_dsn,ref=&ETLSOURCE (&msg),
617 ctl_ds=&dclib..mpe_lockanytable
618 )
619 %mp_lockanytable(UNLOCK
620 ,lib=%scan(&outds_audit,1,.)
621 ,ds=%scan(&outds_audit,2,.)
622 ,ref=&ETLSOURCE
623 ,ctl_ds=&dclib..mpe_lockanytable
624 )
625 %mp_abort(msg=&msg,mac=bitemporal_dataloader.sas);
626 %end;
627%end;
628
629
630/**
631* extract from BASE table. Only want matching records, as could be very BIG.
632* New records are subsequently identified via left join and test for nulls.
633*/
634%local temp_table temp_table2 base_table baselib_schema;
635%put DCNOTE: Extracting matching observations from &base_lib..&base_dsn;
636
637%if &engine_type=OLEDB %then %do;
638 %let temp_table=##%mf_getuniquefileref(prefix=BTMP);
639 %if &loadtype=BITEMPORAL or &loadtype=TXTEMPORAL %then
640 %let base_table=(select * from [dbo].&base_dsn
641 where convert(datetime,&SQLNOW) < &tech_to );
642 %else %let base_table=[dbo].&base_dsn;
643 proc sql;
644 create table &base_lib.."&temp_table"n as
645 select * from work.bitemp0_append;
646 /* open up a connection for pass through SQL */
647 %dc_assignlib(WRITE,&base_lib,passthru=myAlias)
648 create table work.bitemp0_base as select * from connection to myAlias(
649%end;
650%else %if &engine_type=REDSHIFT or &engine_type=POSTGRES or &engine_type=SNOW
651or &engine_type=SASIOSNF
652%then %do;
653 /* grab schema */
654 %let baselib_schema=%mf_getschema(&base_lib);
655 %if &baselib_schema.X ne X %then %let baselib_schema=&baselib_schema..;
656
657 /* grab redshift config */
658 %local redcnt; %let redcnt=0;
659 %if &engine_type=REDSHIFT %then %do;
660 data _null_;
661 set &config_table(where=(var_scope='DCBL_REDSH' and var_active=1));
662 x+1;
663 call symputx(cats('rednm',x),var_value,'l');
664 call symputx(cats('redval',x),var_value,'l');
665 call symputx('redcnt',x,'l');
666 run;
667 %end;
668 %let temp_table=%upcase(%mf_getuniquename(prefix=XDCTEMP));
669 %if &loadtype=BITEMPORAL or &loadtype=TXTEMPORAL %then
670 %let base_table=(select * from &baselib_schema.&base_dsn
671 where timestamp &sqlnow < &tech_to );
672 %else %let base_table=&baselib_schema.&base_dsn;
673 /* make in-db empty table with PK + MD5 only */
674 %dc_assignlib(WRITE,&base_lib,passthru=myAlias)
675 %if &engine_type=SNOW or &engine_type=SASIOSNF %then %do;
676 exec (create transient table &baselib_schema.&temp_table
677 like &baselib_schema.&base_dsn
678 ) by myAlias;
679 %end;
680 %else %do;
681 /* cannot persist temp tables so must create a temporary permanent table */
682 exec (create table &temp_table (like &baselib_schema.&base_dsn)) by myAlias;
683 %if &engine_type=REDSHIFT %then %do;
684 exec (alter table &temp_table alter sortkey none) by myAlias;
685 %end;
686 %end;
687 %local dropcols;
688 %let dropcols=%mf_wordsinstr1butnotstr2(
689 str1=%upcase(%mf_getvarlist(&basecopy))
690 ,str2=%upcase(&pk)
691 );
692 %if %length(&dropcols>0) %then %do idx_pk=1 %to %sysfunc(countw(&dropcols));
693 %put &=dropcols;
694 %let idx_val=%scan(&dropcols,&idx_pk);
695 exec(alter table &temp_table drop column &idx_val;) by myAlias;
696 %end;
697 exec (alter table &temp_table add column &md5_col varchar(32);) by myAlias;
698 /* create view to strip formats and avoid warns in log */
699 data work.vw_bitemp0/view=work.vw_bitemp0;
700 /* inherit remote length to handle byte expansion */
701 if 0 then set &base_lib..&temp_table(keep=&md5_col);
702 set work.bitemp0_append(keep=&pk &md5_col);
703 format _all_;
704 run;
705
706 proc append base=&base_lib..&temp_table
707 %if &engine_type=REDSHIFT %then %do;
708 (
709 %do idx_pk=1 %to &redcnt;
710 &&rednm&idx_pk = &&redval&idxpk
711 %end;
712 )
713 %end;
714 data=work.vw_bitemp0 force nowarn;
715 run;
716 /* open up a connection for pass through SQL */
717 %dc_assignlib(WRITE,&base_lib,passthru=myAlias)
718 create table work.bitemp0_base as select * from connection to myAlias(
719%end;
720%else %if &engine_type=CAS %then %do;
721 %if &loadtype=BITEMPORAL or &loadtype=TXTEMPORAL %then
722 %let base_table=&base_lib..&base_dsn
723 (where=(&tech_from <=&now and &now < &tech_to));
724 %else %let base_table=&base_lib..&base_dsn;
725 %let temp_table=CASUSER.%mf_getuniquename(prefix=DC);
726 data &temp_table;
727 set work.bitemp0_append;
728 run;
729 %let bitemp0base=CASUSER.%mf_getuniquename(prefix=DC);
730 proc fedsql sessref=dcsession;
731 create table &bitemp0base{options replace=true} as
732%end;
733%else %do;
734 %let temp_table=work.bitemp0_append;
735 %if &loadtype=BITEMPORAL or &loadtype=TXTEMPORAL %then
736 %let base_table=&base_lib..&base_dsn
737 (where=(&tech_from <=&now and &now < &tech_to));
738 %else %let base_table=&base_lib..&base_dsn;
739 proc sql;
740 create table work.bitemp0_base as
741%end;
742
743 select a.&md5_col /* this identifies NEW records */
744 , b.*
745 /* assume first PK field cannot be null (if defined in a PK constraint then
746 it definitely cannot be null) */
747 , case when b.%scan(&pk,1) IS NULL then 1 else 0 end as ___TMP___NEW_FLG
748 from &baselib_schema.&temp_table a
749 left join &base_table b
750 on 1=1
751%do idx_pk=1 %to &pk_cnt;
752 %let idx_val=%scan(&pk,&idx_pk);
753 and a.&idx_val=b.&idx_val
754%end;
755
756
757%if &engine_type=OLEDB or &engine_type=REDSHIFT or &engine_type=POSTGRES
758or &engine_type=SNOW or &engine_type=SASIOSNF
759%then %do;
760 ); proc sql; drop table &base_lib.."&temp_table"n;
761%end;
762%else %if &engine_type=CAS %then %do;
763 ;
764 quit;
765 data work.bitemp0_base;
766 set &bitemp0base;
767 run;
768 proc sql;
769 drop table &temp_table;
770 drop table &bitemp0base;
771%end;
772%else %do;
773 ;
774%end;
775
776/**
777* matching & changed records are those without NULL key values
778* &idx_val resolves to rightmost PK value (loop above)
779*/
780%put syscc (line525)=&syscc, sqlrc=&sqlrc;
781%mp_abort(iftrue= (&syscc gt 0 or &sqlrc>0)
782 ,mac=&_program
783 ,msg=%str(syscc=&syscc sqlrc=&sqlrc)
784)
785
786%put hash_char_vars=&hash_char_vars;
787%put hash_num_vars=&hash_num_vars;
788data work.bitemp1_current(drop=___TMP___NEW_FLG);
789 set work.bitemp0_base(drop=&md5_col);
790 if ___TMP___NEW_FLG=0;
791 length &md5_col $32;
792 %mp_rowhash(md5_col=&md5_col
793 ,cvars=&hash_char_vars
794 ,nvars=&hash_num_vars
795 )
796run;
797
798/**
799* NEW records were identified in ___TMP___NEW_FLG in bitemp0_base
800*/
801proc sql;
802create table &outds_add
803 (drop=&md5_col
804 %if %mf_existvar(work.bitemp0_base, &delete_col) %then %do;
805 &delete_col
806 %end;
807 )
808 as select a.*
809 %if &loadtype=BITEMPORAL or &loadtype=TXTEMPORAL %then %do;
810 ,&now as &tech_from &tech_from_fmt
811 ,&high_date as &tech_to &tech_to_fmt
812 %end;
813 from work.bitemp0_append a /* STAGING records (mix of existing & new) */
814 , work.bitemp0_base b /* BASE records (contains null values for new) */
815 where a.&md5_col=b.&md5_col /* took staging md5 across in left join */
816 and b.___TMP___NEW_FLG=1; /* NEW records also identified in bitemp0_base */
817
818
819/**
820* identify INSERTS. These are records with the same business key but
821* the bus_from and bus_to value are higher / lower (respectively)
822* such that the existing record needs to be SPLIT to surround the new
823* record.
824* eg: OLD RECORD from=1 to=10
825* NEW RECORD from=5 to=7
826*
827* APPENDED RECORDS:
828* - from=1 to=5
829* - from=5 to=7
830* - from=7 to=10
831*/
832
833/* inserts cannot happen with TXTEMPORAL */
834%if &loadtype=BITEMPORAL or &loadtype=BUSTEMPORAL %then %do;
835 /* IDENTIFY */
836 create table work.bitemp3_inserts as
837 select b.*
838 ,a.&bus_from as ___TMP___from
839 ,a.&bus_to as ___TMP___to
840 from work.bitemp0_append a
841 ,work.bitemp1_current b
842 where a.&bus_from > b.&bus_from
843 and a.&bus_to < b.&bus_to
844 %do idx_pk=1 %to &pk_cnt;
845 %let idx_val=%scan(&pk,&idx_pk);
846 and a.&idx_val=b.&idx_val
847 %end;
848 order by
849 /* compress blanks and then insert commas (as the datetime fields may
850 not be in use) */
851 %sysfunc(tranwrd(%sysfunc(compbl(
852 &pk &bus_from &bus_to &processed
853 )),%str( ), %str(,)))
854 ;
855
856 /* SPLIT */
857 data work.bitemp3a_inserts (drop=___TMP___from ___TMP___retain ___TMP___to) ;
858 set work.bitemp3_inserts;
859 by &pk &bus_from &bus_to &processed;
860 if first.&idx_val then do;
861 ___TMP___retain=&bus_to;
862 &bus_to=___TMP___from;
863 output;
864 &bus_to=___TMP___retain;
865 end;
866 if last.&idx_val then do;
867 &bus_from=___TMP___to;
868 output;
869 end;
870 run;
871%end;
872%else %do;
873 /* TX temporal load */
874 data work.bitemp3a_inserts;
875 set work.bitemp1_current;
876 stop;
877 run;
878%end;
879/* APPEND */
880proc sql;
881create view work.bitemp3a_view as
882 select * from work.bitemp1_current
883 where &md5_col not in (select &md5_col from work.bitemp3a_inserts);
884
885data bitemp3b_newbase;
886 set work.bitemp3a_inserts work.bitemp3a_view;
887run;
888
889/** do not use! this converts short numerics into 8 bytes
890proc sql;
891create table work.bitemp3b_newbase as
892 select * from work.bitemp3a_inserts
893union corr
894 select * from work.bitemp1_current
895 where &md5_col not in (select &md5_col from work.bitemp3a_inserts);
896*/
897
898/**
899* identify CHANGED records from staging.
900* Same business key with different temporal dates or md5 value
901* This table must be overlayed onto / into existing business history
902*/
903proc sql;
904create table work.bitemp4_updated as select distinct a.*
905 from work.bitemp0_append a
906 ,work.bitemp3b_newbase b
907 where 1=1
908 %do idx_pk=1 %to &pk_cnt;
909 %let idx_val=%scan(&pk,&idx_pk);
910 and a.&idx_val=b.&idx_val
911 %end;
912 and ( a.&md5_col ne b.&md5_col
913 %if &loadtype=BITEMPORAL or &loadtype=BUSTEMPORAL %then %do;
914 OR (a.&bus_from ne b.&bus_from or a.&bus_to ne b.&bus_to)
915 %end;
916 )
917;
918
919/**
920 * This section would have been one simple step with union all
921 * but that converts short numerics into 8 bytes!
922 * so, convoluted alternative to retain the same functionality.
923 */
924
925/* base records */
926create view work.bitemp4_prep1 as
927 select 'BASE' as ___TMP___
928 ,b.*
929 from work.bitemp4_updated a
930 ,work.bitemp3b_newbase b
931 where 1
932 %do idx_pk=1 %to &pk_cnt;
933 %let idx_val=%scan(&pk,&idx_pk);
934 and a.&idx_val=b.&idx_val
935 %end;
936 ;
937/* updated records */
938create view work.bitemp4_prep2 as
939 select 'STAG' as ___TMP___ ,*
940 from work.bitemp4_updated;
941/* ensure we only keep columns that appear in both */
942%local bp1 bp2 bp3 bp4;
943%let bp1=%mf_getvarlist(bitemp4_prep1);
944%let bp2=%mf_getvarlist(bitemp4_prep2);
945%let bp3=%mf_wordsInStr1ButNotStr2(Str1=&bp1,Str2=&bp2);
946%let bp4=%mf_wordsInStr1ButNotStr2(Str1=&bp2,Str2=&bp1);
947data work.bitemp4_prep3/view=bitemp4_prep3;
948 set bitemp4_prep1 bitemp4_prep2;
949%if %length(XX&bp3&bp4)>2 %then %do;
950 drop &bp3 &bp4 ;
951%end;
952run;
953/* remove duplicates */
954proc sql;
955create table work.bitemp4a_allrecs as
956 select distinct *
957 from work.bitemp4_prep3
958 order by
959 /* compress blanks and then insert commas (as the datetime fields
960 may not be in use) */
961 %sysfunc(tranwrd(%sysfunc(compbl(
962 &pk &bus_from &bus_to &processed
963 )),%str( ), %str(,)))
964 ;
965
966%if &loadtype=BITEMPORAL or &loadtype=BUSTEMPORAL %then %do;
967 /* this section aligns the business dates
968 (eg for inserts or overlaps in the range) */
969 data work.bitemp4b_firstpass (drop=___TMP___cond ___TMP___from ___TMP___to );
970 set work.bitemp4a_allrecs;
971 by &pk &bus_from &bus_to &processed;
972 retain ___TMP___cond 'Name of Condition';
973 retain ___TMP___from ___TMP___to 0;
974 ___TMP___md5lag=lag(&md5_col);
975 /* reset retained variables */
976 if first.&idx_val then do;
977 call missing (___TMP___cond, ___TMP___from, ___TMP___to,___TMP___md5lag);
978 end;
979 else do;
980 /* if record is identical, carry forward bus_from (and bus_to if higher)*/
981 if &md5_col=___TMP___md5lag then do;
982 &bus_from=___TMP___from;
983 if &bus_to<___TMP___to then &bus_to=___TMP___to;
984 end;
985 end;
986
987 if ___TMP___='STAG' then do;
988 /* need to carry forward the closing record */
989 ___TMP___cond='Condition 1';
990 end;
991 else if ___TMP___cond='Condition 1' then do;
992 /* else ensure bus_from starts from prior record bus_to */
993 if &md5_col ne ___TMP___md5lag and &bus_from <= ___TMP___to
994 then &bus_from= ___TMP___to;
995 /* new record may replace old record entirely */
996 if &bus_to <= &bus_from then delete;
997 else call missing (___TMP___cond, ___TMP___from, ___TMP___to);
998 end;
999 ___TMP___from=&bus_from;
1000 ___TMP___to=&bus_to;
1001 run;
1002%end;
1003%else %do;
1004 /* keep staged records only */
1005 data work.bitemp4b_firstpass;
1006 set work.bitemp4a_allrecs;
1007 if ___TMP___='STAG';
1008 run;
1009%end;
1010
1011/* next phase is to pass through in reverse - so set up the sort statement */
1012%local byvar;
1013%do idx_pk=1 %to &pk_cnt;
1014 %let byvar=&byvar descending %scan(&pk,&idx_pk);
1015%end;
1016%if &loadtype=BITEMPORAL or &loadtype=BUSTEMPORAL
1017%then %let byvar=&byvar descending &bus_from descending &bus_to;
1018/* if matching bus dates supplied, need to ensure we also have a sort
1019 between BASE and STAGING tables */
1020%let byvar=&byvar descending ___TMP___;
1021
1022proc sort data=work.bitemp4b_firstpass out=work.bitemp4c_sort ;
1023 by &byvar;
1024run;
1025
1026/**
1027* Now (in reverse) pass back business start dates
1028*/
1029data work.bitemp4d_secondpass;
1030%if &loadtype=BITEMPORAL or &loadtype=TXTEMPORAL %then %do;
1031 &tech_from=&now;
1032 &tech_to=&high_date;
1033%end;
1034 set work.bitemp4c_sort ;
1035 by &byvar;
1036 retain ___TMP___cond 'Name of Condition';
1037 retain ___TMP___from ___TMP___to 0;
1038%if &loadtype=BITEMPORAL or &loadtype=BUSTEMPORAL %then %do;
1039/* put / _all_ /;*/
1040 ___TMP___md5lag=lag(&md5_col);
1041 if first.&idx_val then do;
1042 /* reset retained variables */
1043 call missing (___TMP___cond,___TMP___from,___TMP___to,___TMP___md5lag);
1044 end;
1045 else do;
1046 /* if record is identical, carry back bus_to */
1047 if &md5_col=___TMP___md5lag then &bus_to=___TMP___to;
1048 end;
1049
1050 if ___TMP___='STAG' then do;
1051 /* need to carry forward the closing record */
1052 ___TMP___cond='Condition 2';
1053 end;
1054 else if ___TMP___cond='Condition 2' then do;
1055 /* else ensure bus_to stops at subsequent record bus_from */
1056 if &md5_col ne ___TMP___md5lag and &bus_to >= ___TMP___from
1057 then &bus_to= ___TMP___from;
1058 /* new record may replace old record entirely */
1059 if &bus_from >= &bus_to then delete;
1060 if &bus_from=___TMP___from and &bus_to=___TMP___to then delete;
1061 else call missing (___TMP___cond, ___TMP___from, ___TMP___to);
1062 end;
1063 ___TMP___from=&bus_from;
1064 ___TMP___to=&bus_to;
1065
1066%end;
1067run;
1068%put syscc (line600)=&syscc;
1069/**
1070 There may still be some records (eg old business history) which have not
1071 changed.
1072 Need to identify these and remove from the append so they are not updated
1073 unnecessarily. This is done by generating a new md5 (which INCLUDES the
1074 business key) and any matching / identical records are split out (from those
1075 that need to be updated).
1076*/
1077
1078%if &loadtype=BITEMPORAL %then %do;
1079 /* For bitemporal we also include the business dates in the comparison hash.
1080 The business dates are passed as prefix numerics so that both lookup and
1081 update tables use exactly the same hashing order. */
1082
1083 data work.bitemp5a_lkp (keep=&md5_col)
1084 %if "%substr(&sysver,1,1)" ne "4" & "%substr(&sysver,1,1)" ne "5" %then %do;
1085 /nonote2err
1086 %end;
1087 ;
1088 set work.bitemp0_base;
1089 %mp_rowhash(md5_col=&md5_col
1090 ,cvars=&hash_char_vars
1091 ,nvars=&bus_from &bus_to &hash_num_vars
1092 )
1093 run;
1094
1095 data bitemp5b_updates;
1096 set bitemp4d_secondpass;
1097 if _n_=1 then do;
1098 dcl hash md5_lkp(dataset:'bitemp5a_lkp');
1099 md5_lkp.definekey("&md5_col");
1100 md5_lkp.definedone();
1101 end;
1102 /* drop old md5 col as will rebuild with new business dates */
1103 %mp_rowhash(md5_col=&md5_col
1104 ,cvars=&hash_char_vars
1105 ,nvars=&bus_from &bus_to &hash_num_vars
1106 )
1107 if md5_lkp.check()=0 then delete;
1108 run;
1109
1110 proc sql;
1111 /* get min bus from as will update (close out) all records from this point
1112 (for that PK)*/
1113 create table work.bitemp5d_subquery as
1114 select &pk_comma, min(&bus_from)as &bus_from, max(&bus_to) as &bus_to
1115 from work.bitemp5b_updates
1116 group by &pk_comma;
1117 /* index has a huge efficiency impact on upcoming nested subquery */
1118 create index index1 on work.bitemp5d_subquery(&pk_comma,&bus_from, &bus_to);
1119
1120 %let lastds=work.bitemp5b_updates;
1121%end;
1122%else %if &loadtype=TXTEMPORAL or &loadtype=UPDATE %then %do;
1123 proc sql;
1124 create table work.bitemp5d_subquery as
1125 select distinct &pk_comma
1126 from bitemp4d_secondpass;
1127 %let lastds=work.bitemp4d_secondpass;
1128%end;
1129%else %let lastds=work.bitemp4d_secondpass;
1130
1131/* create single append table (an overlapped pre-sert may be classed as
1132 both an update AND a new record). Also create temp views that may be
1133 used for pre-load analysis. */
1134data &outds_mod;
1135 set &lastds(drop=___TMP___: &md5_col);
1136run;
1137
1138data bitemp6_allrecs / view=bitemp6_allrecs;
1139 set &outds_mod /* UPDATED records */
1140 &outds_add /* NEW records */;
1141run;
1142
1143proc sort data=work.bitemp6_allrecs
1144 out=work.bitemp6_unique
1145 noduprec
1146 dupout=work.xx_BADBADBAD;
1147by _all_;
1148run;
1149
1150/* we have all our temp tables now so exit if this is all that is needed */
1151%if &LOADTARGET ne YES %then %return;
1152
1153/* also exit if an err condition exists */
1154
1155%if &syscc>0 %then %do;
1156 %put syscc=&syscc;
1157 %mp_lockanytable(UNLOCK,lib=&base_lib,ds=&base_dsn,ref=&ETLSOURCE,
1158 ctl_ds=&dclib..mpe_lockanytable
1159 )
1160 %if "&outds_audit" ne "0" %then %do;
1161 %mp_lockanytable(UNLOCK
1162 ,lib=%scan(&outds_audit,1,.)
1163 ,ds=%scan(&outds_audit,2,.)
1164 ,ref=&ETLSOURCE
1165 ,ctl_ds=&dclib..mpe_lockanytable
1166 )
1167 %end;
1168%end;
1169%mp_abort(iftrue= (&syscc>0)
1170 ,mac=&sysmacroname in &_program
1171 ,msg=%str(Bitemporal transform / job aborted due to SYSCC=&SYSCC status)
1172)
1173
1174/* final check - abort if a lock has appeared on the target or audit table */
1175%mp_lockfilecheck(libds=&base_lib..&base_dsn)
1176%if %mf_existds(&outds_audit) %then %do;
1177 %mp_lockfilecheck(libds=&outds_audit)
1178%end;
1179
1180/**
1181* STAGING TABLES PREPARED, ERR CONDITION TESTED FOR.. NOW TO LOAD!!
1182*/
1183
1184/**
1185* First, CLOSE OUT changed records (if not a REPLACE)
1186* Note that SAS does not support ANSI standard for UPDATE with a join condition.
1187* However - this can be worked around using a nested subquery..
1188*/
1189data _null_;
1190 putlog "&sysmacroname: CLOSEOUTS commencing";
1191run;
1192
1193%if %mf_getattrn(&lastds,NLOBS)=0 %then %do;
1194 data _null_;
1195 putlog "&sysmacroname: No closeouts needed";
1196 run;
1197%end;
1198%else %if &engine_type=CAS %then %do;
1199 %mp_abort(iftrue= (&loadtype=BITEMPORAL or &loadtype=TXTEMPORAL)
1200 ,mac=&sysmacroname in &_program
1201 ,msg=%str(&loadtype not yet supported in CAS engine)
1202 )
1203 /* create temp table for deletions */
1204 %local delds;%let delds=%mf_getuniquename(prefix=DC);
1205 data casuser.&delds;
1206 set work.bitemp5d_subquery;
1207 run;
1208 /* delete the records */
1209 proc cas ;
1210 table.deleteRows / table={
1211 caslib="&base_lib",
1212 name="&base_dsn",
1213 where="1=1",
1214 whereTable={caslib='CASUSER',name="&delds"}
1215 };
1216 quit;
1217 /* drop temp table */
1218 proc sql;
1219 drop table CASUSER.&delds;
1220%end;
1221%else %if (&loadtype=BITEMPORAL or &loadtype=TXTEMPORAL or &loadtype=UPDATE)
1222%then %do;
1223 data _null_;
1224 putlog "&sysmacroname: &loadtype operation using *&engine_type* engine";
1225 run;
1226 %local flexinow;
1227 proc sql;
1228 /* if OLEDB then create a temp table for efficiency */
1229 %local innertable;
1230 %if &engine_type=OLEDB %then %do;
1231 %let innertable=[&temp_table];
1232 %let top_table=[dbo].&base_dsn;
1233 %let flexinow=&SQLNOW;
1234 create table &base_lib.."&temp_table"n as
1235 select * from work.bitemp5d_subquery;
1236 /* open up a connection for pass through SQL */
1237 %dc_assignlib(WRITE,&base_lib,passthru=myAlias)
1238 execute(
1239 %end;
1240 %else %if &engine_type=REDSHIFT or &engine_type=POSTGRES or &engine_type=SNOW
1241 or &engine_type=SASIOSNF
1242 %then %do;
1243 %let innertable=%upcase(%mf_getuniquename(prefix=XDCTEMP));
1244 %let top_table=&baselib_schema.&base_dsn;
1245 %let flexinow=timestamp &SQLNOW;
1246 /* make empty table first - must clone & drop extra cols
1247 as autoload is bad */
1248 %dc_assignlib(WRITE,&base_lib,passthru=myAlias)
1249 %if &engine_type=SNOW or &engine_type=SASIOSNF %then %do;
1250 exec (create transient table &baselib_schema.&innertable
1251 like &baselib_schema.&base_dsn
1252 ) by myAlias;
1253 %end;
1254 %else %do;
1255 exec (create table &innertable
1256 (like &baselib_schema.&base_dsn)
1257 ) by myAlias;
1258 %if &engine_type=REDSHIFT %then %do;
1259 exec (alter table &innertable alter sortkey none) by myAlias;
1260 %end;
1261 %end;
1262 %let dropcols=%mf_wordsinstr1butnotstr2(
1263 str1=%upcase(%mf_getvarlist(&basecopy))
1264 ,str2=%upcase(%mf_getvarlist(work.bitemp5d_subquery))
1265 );
1266 %if %length(&dropcols>0) %then %do idx_pk=1 %to %sysfunc(countw(&dropcols));
1267 %put &=dropcols;
1268 %let idx_val=%scan(&dropcols,&idx_pk);
1269 exec(alter table &innertable drop column &idx_val;) by myAlias;;
1270 %end;
1271 /* create view to strip formats and avoid warns in log */
1272 data work.vw_bitemp5d/view=work.vw_bitemp5d;
1273 set work.bitemp5d_subquery;
1274 format _all_;
1275 run;
1276 proc append base=&base_lib..&innertable (
1277 %do idx_pk=1 %to &redcnt;
1278 &&rednm&idx_pk = &&redval&idxpk
1279 %end;
1280 )
1281 data=work.vw_bitemp5d force nowarn;
1282 run;
1283 /* open up a connection for pass through SQL */
1284 %dc_assignlib(WRITE,&base_lib,passthru=myAlias)
1285 execute(
1286 %end;
1287 %else %do;
1288 %put Not using passthrough for *&engine_type* engine;
1289 %let innertable=bitemp5d_subquery;
1290 %let top_table=&base_lib..&base_dsn;
1291 %let flexinow=&now;
1292 %end;
1293
1294
1295 %if &loadtype=BITEMPORAL or &loadtype=TXTEMPORAL %then %do;
1296 update &top_table set &tech_to=&flexinow
1297 %if %length(&processed)>0 %then %do;
1298 ,&processed=&flexinow
1299 %end;
1300 where &tech_from <= &flexinow and &flexinow < &tech_to and
1301 %end;
1302 %else %if &loadtype=UPDATE %then %do;
1303 /* changed records are deleted then re-appended when doing UPDATEs */
1304 delete from &top_table where
1305 %end;
1306 %else %do;
1307 %put %str(ERR)OR: BUSTEMPORAL NOT YET SUPPORTED;
1308 %let syscc=5;
1309 %mp_lockanytable(UNLOCK,lib=&base_lib,ds=&base_dsn,ref=&ETLSOURCE,
1310 ctl_ds=&dclib..mpe_lockanytable
1311 )
1312 %mp_lockanytable(UNLOCK
1313 ,lib=%scan(&outds_audit,1,.)
1314 ,ds=%scan(&outds_audit,2,.)
1315 ,ref=&ETLSOURCE
1316 ,ctl_ds=&dclib..mpe_lockanytable
1317 )
1318 %goto end_of_macro;
1319 %end;
1320
1321 /* perform join inside query as per
1322 http://stackoverflow.com/questions/24629793/update-with-a-proc-sql */
1323
1324 exists( select 1 from &baselib_schema.&innertable where
1325
1326 /* loop PK join */
1327 %do idx_pk=1 %to &pk_cnt;
1328 %let idx_val=%scan(&pk,&idx_pk);
1329 &base_dsn..&idx_val=&innertable..&idx_val and
1330 %end;
1331 %if &loadtype=BITEMPORAL %then %do;
1332 &base_dsn..&bus_from >= &innertable..&bus_from
1333 and &base_dsn..&bus_to <= &innertable..&bus_to and
1334 %end;
1335
1336 /* close the statement */
1337
1338 1=1);
1339
1340 %if &engine_type=OLEDB or &engine_type=REDSHIFT or &engine_type=POSTGRES
1341 or &engine_type=SNOW or &engine_type=SASIOSNF
1342 %then %do;
1343 ) by myAlias;
1344 execute (drop table &baselib_schema.&innertable) by myAlias;
1345 %end;
1346%end;
1347quit;
1348data _null_;
1349 putlog "&sysmacroname: Closeout complete";
1350run;
1351/**
1352 * Append the new / updated records
1353 */
1354%if &engine_type=CAS %then %do;
1355
1356 /* get varchar variables ready for casting */
1357 %local vcfmt vcrename vcassign vcdrop;
1358 data _null_;
1359 set work.bitemp_cols(where=(type=6)) end=last;
1360 length vcrename vcassign vcdrop vcfmt $32767 rancol $32;
1361 retain vcrename vcassign vcdrop vcfmt;
1362 if _n_=1 then vcrename='(rename=(';
1363 rancol=resolve('%mf_getuniquename()');
1364 vcfmt=trim(vcfmt)!!'length '!!cats(name)!!' varchar(*);';
1365 vcrename=trim(vcrename)!!' '!!cats(name,'=',rancol);
1366 vcassign=cats(vcassign,name,'=',rancol,';');
1367 vcdrop=cats(vcdrop,'drop '!!rancol,';');
1368 if last then do;
1369 vcrename=cats(vcrename,'))');
1370 call symputx('vcfmt',vcfmt);
1371 call symputx('vcrename',vcrename);
1372 call symputx('vcassign',vcassign);
1373 call symputx('vcdrop',vcdrop);
1374 end;
1375 run;
1376
1377 /* prepare a temp cas table with varchars casted */
1378 %let tmp=%mf_getuniquename();
1379 data casuser.&tmp ;
1380 &vcfmt
1381 set work.bitemp6_unique &vcrename;
1382 &vcassign
1383 &vcdrop
1384 run;
1385
1386 /* load the table with varchars applied*/
1387 data &base_lib..&base_dsn (append=yes )/sessref=dcsession ;
1388 set casuser.&tmp;
1389 run;
1390
1391 /* drop temp table */
1392 proc sql;
1393 drop table CASUSER.&tmp;
1394
1395 /* this code will not work as regular tables do not have varchars */
1396 /*
1397 proc casutil;
1398 load data=work.bitemp6_unique
1399 outcaslib="&base_lib" casout="&base_dsn" append ;
1400 quit;
1401 */
1402%end;
1403%else %if &engine_type=REDSHIFT or &engine_type=POSTGRES %then %do;
1404 proc append base=&base_lib..&base_dsn
1405 %if &engine_type=REDSHIFT %then %do;
1406 (
1407 %do idx_pk=1 %to &redcnt;
1408 &&rednm&idx_pk = &&redval&idxpk
1409 %end;
1410 )
1411 %end;
1412 data=bitemp6_unique force nowarn;
1413 run;
1414%end;
1415%else %do;
1416 proc append base=&base_lib..&base_dsn data=bitemp6_unique force nowarn; run;
1417%end;
1418
1419%mp_lockanytable(UNLOCK,lib=&base_lib,ds=&base_dsn,ref=&ETLSOURCE,
1420 ctl_ds=&dclib..mpe_lockanytable
1421)
1422
1423/* final check on syscc */
1424%mp_abort(iftrue= (&syscc >4)
1425 ,mac=&_program
1426 ,msg=%str(!!Upload NOT successful!! Failed on actual update / append stage..)
1427)
1428
1429%if &outds_audit ne 0 and &LOADTARGET=YES %then %do;
1430 data work.vw_outds_orig /view=work.vw_outds_orig;
1431 set work.bitemp0_base (drop=&md5_col);
1432 where ___TMP___NEW_FLG=0;
1433 drop ___TMP___NEW_FLG;
1434 run;
1435 /* update the AUDIT table */
1436 %if %mf_existds(&outds_audit) %then %do;
1437 options mprint;
1438 %mp_storediffs(&base_lib..&base_dsn
1439 ,work.vw_outds_orig
1440 ,&pk &bus_from
1441 ,delds=&outds_del
1442 ,modds=&outds_mod
1443 ,appds=&outds_add
1444 ,outds=work.mp_storediffs
1445 ,processed_dttm=&now
1446 ,loadref=%superq(etlsource)
1447 )
1448 /* exclude unchanged values in modified rows */
1449 data work.mp_storediffs;
1450 set work.mp_storediffs;
1451 if MOVE_TYPE="M" and IS_PK=0 and IS_DIFF=0 then delete;
1452 * putlog load_ref= libref= dsn= key_hash= tgtvar_nm=;
1453 run;
1454 proc append base=&outds_audit data=work.mp_storediffs;
1455 run;
1456 %mp_lockanytable(UNLOCK
1457 ,lib=%scan(&outds_audit,1,.)
1458 ,ds=%scan(&outds_audit,2,.)
1459 ,ref=&ETLSOURCE
1460 ,ctl_ds=&dclib..mpe_lockanytable
1461 )
1462 %end;
1463%end;
1464%mp_abort(iftrue= (&syscc >4)
1465 ,mac=bitemporal_dataloader
1466 ,msg=%str(Problem in audit stage (&outds_audit))
1467)
1468
1469%let user=%mf_getUser();
1470/**
1471 Notify as appropriate EMAILS DISABLED
1472
1473%sumo_alerts(ALERT_EVENT=UPDATE
1474 , ALERT_TARGET=&base_lib..&base_dsn
1475 , from_user= &user);
1476*/
1477/* monitor BiTemporal usage */
1478%if &log=1 %then %do;
1479 %put syscc=&syscc;
1480 /* do not perform duration calc in pass through */
1481 %local dur;
1482 data _null_;
1483 now=symget('now');
1484 dur=%sysfunc(datetime())-&now;
1485 call symputx('dur',dur,'l');
1486 run;
1487 proc sql;
1488 insert into &dclib..mpe_dataloads
1489 set libref=%upcase("&base_lib")
1490 ,DSN=%upcase("&base_dsn")
1491 ,ETLSOURCE="&ETLSOURCE"
1492 ,LOADTYPE="&loadtype"
1493 ,CHANGED_RECORDS=%mf_getattrn(&lastds,NLOBS)
1494 ,NEW_RECORDS=%mf_getattrn(&outds_add,NLOBS)
1495 ,DELETED_RECORDS=%mf_getattrn(&outds_del,NLOBS)
1496 ,DURATION=&dur
1497 ,MAC_VER="v&ver"
1498 ,user_nm="&user"
1499 ,PROCESSED_DTTM=&now;
1500 quit;
1501 %put syscc=&syscc;
1502%end;
1503%end_of_macro:
1504%mend bitemporal_dataloader;