dc/sas/sasjs/macros/mpe_refreshtables.sas

/**
  @file mpe_refreshtables.sas
  @brief Refreshes the data catalog
  @details

  @param [in] etlsource= (0) Defaults to %mf_getuniquename(prefix=refreshtables), identifies this load
    in the MPE_DATALOADS audit row.  That table is keyed on etlsource together
    with processed_dttm, libref and dsn, so the uid keeps each load distinct.
  Usage:

      %mpe_refreshtables(sashelp)

  <h4> SAS Macros </h4>
  @li mf_getuniquename.sas
  @li mp_abort.sas
  @li bitemporal_dataloader.sas
  @li mf_existfeature.sas
  @li mf_getengine.sas
  @li mf_getschema.sas
  @li mp_getconstraints.sas

  @version 9.3
  @author 4GL Apps Ltd
**/

%macro mpe_refreshtables(lib,ds=#all
  ,etlsource=0
);
/* a default of 0 rather than a %mf_uid() expression: a default is
  resolved when the macro is compiled, so it would be one value for the
  whole session and the audit folder below it would collide on the
  second call - as it did between the refresh steps of one run */
%if "&etlsource"="0" %then %let etlsource=%mf_getuniquename(prefix=refreshtables);
%let lib=%upcase(&lib);
%let ds=%upcase(&ds);
%local engine; %let engine=%mf_getengine(&lib);
%local schema; %let schema=%mf_getschema(&lib);
%put running &sysmacroname &lib(&engine &schema) for &ds;
proc sql;
create table cols as
  select  libname as libref
    ,upcase(memname) as dsn
    ,memtype
    ,upcase(name) as name
    ,type
    ,length
    ,varnum
    ,label
    ,format
    ,idxusage
    ,notnull
  from dictionary.columns
  where upcase(libname)="&lib"
%if &ds ne #ALL %then %do;
    and upcase(memname)="&ds"
%end;
  ;

%mp_abort(iftrue= (&syscc ne 0)
  ,mac=&_program
  ,msg=%str(syscc=&syscc afer &lib cols extraction)
)

%if &engine=SQLSVR %then %do;
  proc sql;
  connect using &lib;
  create table work.indexes as
  select * from connection to &lib(
  select
      s.name as SchemaName,
      t.name as memname,
      tc.name as name,
      ic.key_ordinal as KeyOrderNr
  from
      sys.schemas s
      inner join sys.tables t   on s.schema_id=t.schema_id
      inner join sys.indexes i  on t.object_id=i.object_id
      inner join sys.index_columns ic on i.object_id=ic.object_id
                                    and i.index_id=ic.index_id
      inner join sys.columns tc on ic.object_id=tc.object_id
                              and ic.column_id=tc.column_id
  where i.is_primary_key=1
    and s.name=%str(%')&schema%str(%')
  order by t.name, ic.key_ordinal ;
  );disconnect from &lib;
  create table finalcols as
    select a.*
      ,case when b.name is not null then 1 else 0 end as pk_ind
    from work.cols a
    left join work.indexes b
    on a.dsn=b.memname
      and upcase(a.name)=upcase(b.name)
    order by libref,dsn;
%end;
%else %do;
  %local dsn;
  %if &ds = #ALL %then %let dsn=;

  %mp_getconstraints(lib=&lib.,ds=&dsn,outds=work.constraints)


  /* extract cols that are clearly primary keys */
  proc sql;
  create table work.pk4sure as
    select libref
      ,table_name
      ,constraint_name
      ,constraint_order
      ,column_name as name
    from work.constraints
    where constraint_type='PRIMARY'
    order by 1,2,3,4;

  /* extract unique constraints where every col is also NOT NULL */
  proc sql;
  create table work.sum as
    select a.libref
      ,a.table_name
      ,a.constraint_name
      ,count(a.column_name) as unq_cnt
      ,count(b.column_name) as nul_cnt
    from work.constraints(where=(constraint_type ='UNIQUE')) a
    left join work.constraints(where=(constraint_type ='NOT NULL')) b
    on a.libref=b.libref
      and a.table_name=b.table_name
      and a.column_name=b.column_name
    group by 1,2,3
    having unq_cnt=nul_cnt;


  /* extract cols from the relevant unique constraints */
  create table work.pkdefault as
    select a.libref
      ,a.table_name
      ,a.constraint_name
      ,b.constraint_order
      ,b.column_name as name
    from work.sum a
    left join work.constraints(where=(constraint_type ='UNIQUE')) b
    on a.libref=b.libref
      and a.table_name=b.table_name
      and a.constraint_name=b.constraint_name
    order by 1,2,3,4;

  /* extract cols from the relevant unique INDEXES */
  create table work.pkfromindex as
    select libname as libref
      ,memname as table_name
      ,indxname as constraint_name
      ,indxpos as constraint_order
      ,name
    from dictionary.indexes
    where nomiss='yes' and unique='yes' and upcase(libname)="&lib"
  %if &ds ne #ALL %then %do;
      and upcase(memname)="&ds"
  %end;
    order by 1,2,3,4;

  /* create one table */
  data work.finalpks;
    set pkdefault pk4sure pkfromindex;
    pk_ind=1;
    /* if there are multiple unique constraints, take the first */
    by libref table_name constraint_name;
    retain keepme;
    if first.table_name then keepme=1;
    if first.constraint_name and not first.table_name then keepme=0;
    if keepme=1;
  run;

  /* join back to starting table */
  proc sql;
  create table finalcols as
    select a.*
      ,b.constraint_order
      ,case when b.pk_ind=1 then 1 else 0 end as pk_ind
    from work.cols a
    left join work.finalpks b
    on a.libref=b.libref
      and a.dsn=b.table_name
      and upcase(a.name)=upcase(b.name)
    order by libref,dsn,constraint_order;

%end;

%mp_abort(iftrue= (&syscc ne 0)
  ,mac=&_program
  ,msg=%str(syscc=&syscc afer &lib indexes extraction)
)

/* load columns */
%bitemporal_dataloader(base_lib=&mpelib
  ,base_dsn=mpe_datacatalog_vars
  ,append_dsn=finalcols
  ,PK=LIBREF DSN NAME
  ,etlsource=&etlsource
  ,loadtype=TXTEMPORAL
  ,tech_from=TX_FROM
  ,tech_to=TX_TO
%if &ds ne #ALL %then %do;
    ,close_vars=LIBREF DSN
%end;
  ,dclib=&mpelib
)

/* prepare tables */
proc sql;
create table work.tabs as select
  libname as libref
  ,upcase(memname) as dsn
  ,memtype
%if %mf_existfeature(DBMS_MEMTYPE)=1 %then %do;
  ,dbms_memtype
%end;
%else %do;
  ,'n/a' as dbms_memtype format=$32.
%end;
  ,typemem
  ,memlabel
  ,nvar
  ,compress
from dictionary.tables
  where upcase(libname)="&lib"
%if &ds ne #ALL %then %do;
    and upcase(memname)="&ds"
%end;
  ;
data tabs2;
  set finalcols;
  length pk_fields $512;
  retain pk_fields;
  by libref dsn;
  if first.dsn then pk_fields='';
  if pk_ind=1 then pk_fields=catx(' ',pk_fields,name);
  if last.dsn then output;
run;

proc sql;
create table work.finaltabs as
  select a.libref
    ,a.dsn
    ,a.memtype
    ,a.dbms_memtype
    ,a.typemem
    ,a.memlabel
    ,a.nvar
    ,a.compress
    ,b.pk_fields
  from work.tabs a
  left join work.tabs2 b
  on a.libref=b.libref
    and a.dsn=b.dsn;

%bitemporal_dataloader(base_lib=&mpelib
  ,base_dsn=mpe_datacatalog_tabs
  ,append_dsn=finaltabs
  ,PK=LIBREF DSN
  ,etlsource=&etlsource
  ,loadtype=TXTEMPORAL
  ,tech_from=TX_FROM
  ,tech_to=TX_TO
  ,dclib=&mpelib
%if &ds ne #ALL %then %do;
  ,close_vars=LIBREF
%end;
)

/* prepare table frequently changing attributes */
proc sql;
%if &engine=SQLSVR %then %do;
  connect using &lib;
  create table work.attrs as select * from connection to &lib(
    SELECT SCHEMA_NAME(schema_id) as 'schema', name, create_date, modify_date
      FROM sys.tables ;
  );
  create table work.nobs as select * from connection to &lib(
    SELECT      SCHEMA_NAME(A.schema_id) AS 'schema'
        ,A.Name, AVG(B.rows) AS 'RowCount'
      FROM        sys.objects A
      INNER JOIN sys.partitions B ON A.object_id = B.object_id
      WHERE       A.type = 'U'
      GROUP BY    A.schema_id, A.Name
  );
  disconnect from &lib;
  create table statustabs as select
      a.libref
      ,a.dsn
      ,b.create_date as crdate
      ,b.modify_date as modate
      ,. as filesize
      ,c.RowCount as nobs
    from work.tabs a
    left join work.attrs(where=(schema="&schema")) b
    on upcase(a.dsn)=upcase(b.name)
    left join work.nobs(where=(schema="&schema")) c
    on upcase(a.dsn)=upcase(c.name);
%end;
%else %do;
  create table statustabs as select
      libname as libref
      ,upcase(memname) as dsn
      ,crdate
      ,modate
      ,filesize
      ,nobs
  from dictionary.tables
    where upcase(libname)="&lib"
  %if &ds ne #ALL %then %do;
      and upcase(memname)="&ds"
  %end;
  ;
%end;

%bitemporal_dataloader(base_lib=&mpelib
  ,base_dsn=mpe_datastatus_tabs
  ,append_dsn=statustabs
  ,PK=LIBREF DSN
  ,etlsource=&etlsource
  ,loadtype=TXTEMPORAL
  ,tech_from=TX_FROM
  ,tech_to=TX_TO
  ,dclib=&mpelib
%if &ds ne #ALL %then %do;
  ,close_vars=LIBREF
%end;
)

%if &ds = #ALL %then %do;
  proc sql;
  create table work.sumcat as
    select libname as libref,
      count(distinct memname) as catalog_cnt
    from dictionary.catalogs
    where upcase(libname)="&lib"
    group by 1;
  create table work.sumdsn as select
      libref
      ,sum(filesize) as libsize
      ,count(*) as table_cnt
    from statustabs
    group by 1;
  create table work.libs as
    select libref from work.sumcat
    union
    select libref from work.sumdsn;
  create table work.statuslibs as
    select a.libref,
      b.libsize,
      b.table_cnt,
      c.catalog_cnt
    from work.libs a
    left join work.sumdsn b
    on a.libref=b.libref
    left join work.sumcat c
    on a.libref=c.libref;

  %bitemporal_dataloader(base_lib=&mpelib
    ,base_dsn=mpe_datastatus_libs
    ,append_dsn=statuslibs
    ,PK=LIBREF
    ,etlsource=&etlsource
    ,loadtype=TXTEMPORAL
    ,tech_from=TX_FROM
    ,tech_to=TX_TO
    ,dclib=&mpelib
  )
%end;

%mend mpe_refreshtables;