dc/dc
7
1
Fork 0
Code Issues 19 Pull Requests 3 Actions 2 Packages Projects 1 Releases 91 Wiki Activity

Merge branch 'version7-13' into additional-validations-formulae
Build / Build-and-ng-test (pull_request) Failing after 2m2s
Details
Build / Build-and-test-development (pull_request) Skipped
Details
Lighthouse Checks / lighthouse (pull_request) Successful in 21m21s
Details

Browse Source
This commit is contained in:
Yury
2026-08-04 08:19:03 +00:00
parent a8237b2881 28104f83e1
commit b88b22684a
5 changed files with 591 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.agent/docs/bitemporal-dataloader.md
+147
View File
@@ -0,0 +1,147 @@
# Bitemporal Dataloader — Technical Deep Dive
This document explains how the Data Controller backend loads staged data into target tables: the load-type dispatch, the internals of the `%bitemporal_dataloader` macro, and a detailed description of the REPLACE load type. For user-facing load type documentation, see [docs.datacontroller.io](https://docs.datacontroller.io/).
## Overview
Every table registered for loading has a current record in `&mpelib..MPE_TABLES` with a `LOADTYPE` (selectbox values are seeded in `mpe_makedata.sas`):
| LOADTYPE | Loader used | History kept |
|---|---|---|
| `UPDATE` | `%bitemporal_dataloader` (no temporal vars) | None — changed records are deleted and re-appended |
| `REPLACE` | Inline code in `%mpe_targetloader` (does **not** use `%bitemporal_dataloader`) | None — entire table wiped and reloaded |
| `TXTEMPORAL` | `%bitemporal_dataloader` (technical time only) | SCD2-style, technical (transaction) time |
| `BITEMPORAL` | `%bitemporal_dataloader` (business + technical time) | Full two-dimensional history |
| `FORMAT_CAT` | `%mp_loadformat` | Format catalog load (table suffix `-FC`) |
The relevant `MPE_TABLES` columns read by the loader are: `LOADTYPE`, `BUSKEY` (primary key, space-separated, excluding temporal columns), `VAR_TXFROM` / `VAR_TXTO` (technical validity), `VAR_BUSFROM` / `VAR_BUSTO` (business validity), `VAR_PROCESSED` (processed timestamp column), `RK_UNDERLYING` (retained-key generation), `CLOSE_VARS`, and `AUDIT_LIBDS` (defaults to `&dclib..MPE_AUDIT`).
## Request Flow
```mermaid
flowchart TD
A[User submits changeset\neditors/stagedata.sas] --> B[Staging package written to\n&mpelocapprovals/&LOAD_REF\nCSV + jsdata]
B --> C[Approval workflow\nMPE_SUBMIT / MPE_REVIEW]
C --> D{auditors/postdata.sas}
D -->|action=SHOW_DIFFS| E["%mpe_targetloader(LOADTARGET=NO)\nbuilds work.outds_add / outds_mod / outds_del\nfor the diff screen only"]
D -->|action=APPROVE_TABLE| F["%mpe_targetloader(LOADTARGET=YES)\nactual load"]
E --> G[Diff CSV + TEMPDIFFS stored\nin approval package]
F --> H{LOADTYPE from\nMPE_TABLES}
H -->|UPDATE / TXTEMPORAL / BITEMPORAL| I["%bitemporal_dataloader"]
H -->|FORMAT_CAT| J["%mp_loadformat"]
H -->|REPLACE| K[Inline delete-all + append\nin %mpe_targetloader]
```
`%mpe_targetloader` (`sas/sasjs/macros/mpe_targetloader.sas`) is the single dispatch point. It reads the current `MPE_TABLES` record (`&dc_dttmtfmt. lt tx_to`), aborts if the table is not registered (or has duplicate config records), and routes to the loader. Note the two-phase design: `LOADTARGET=NO` prepares the intermediate `outds_*` tables so the approver can review diffs; `LOADTARGET=YES` performs the destructive load. Both phases run the same preparation logic, so the reviewed diffs correspond to what is actually applied.
## The Temporal Model
Bitemporal tables carry two independent time dimensions:
* **Business time** (`bus_from` / `bus_to`) — when the fact was true in the real world. Present on **both** staging and base tables.
* **Technical time** (`tech_from` / `tech_to`, a.k.a. transaction time) — when the record was known to the database. Present on the **base table only**; the loader stamps these itself.
All validity is expressed with **half-open intervals** (`from <= t < to`). Queries against bitemporal tables need two conditions and must not use `BETWEEN` or `from LE t LE to` (the latter excludes boundary records — see the macro header for background):
```sas
where &bus_from le [tstamp] lt &bus_to
and &tx_from le [tstamp] lt &tx_to
```
"Current" records have `tech_to` set to the high date (`'31DEC9999:23:59:59'dt` when called from `%mpe_targetloader`). Closing out a record means setting `tech_to = now` — records are never physically deleted from a temporal table, they are superseded.
## Key Components
| Component | Location | Role |
|---|---|---|
| `%mpe_targetloader` | `sas/sasjs/macros/mpe_targetloader.sas` | Reads `MPE_TABLES` config and dispatches per LOADTYPE; implements REPLACE inline |
| `%bitemporal_dataloader` | `sas/sasjs/macros/bitemporal_dataloader.sas` | Generic loader for UPDATE / TXTEMPORAL / BITEMPORAL |
| `%bitemporal_closeouts` | `sas/sasjs/macros/bitemporal_closeouts.sas` | Closes out (sets `tech_to=now`) live records matching a key |
| `%mp_retainedkey` | SASjs core | Generates retained (surrogate) keys when `RK_UNDERLYING` is configured |
| `%mp_rowhash` | SASjs core | MD5 hash of non-temporal columns, used for change detection |
| `%mp_storediffs` | SASjs core | Writes row-level audit records to `AUDIT_LIBDS` |
| `MPE_DATALOADS` | `&dclib` | Load log (counts, duration, macro version, user) — written only by `%bitemporal_dataloader` |
| `MPE_LOCKANYTABLE` | `&dclib` | Lock control table used by `%mp_lockanytable` |
## `%bitemporal_dataloader` Execution Flow
The macro builds a series of `work.bitemp*` intermediate tables, then applies the changes in two target-table operations (closeout, then append) under a single lock.
### 1. Pre-checks and setup
* Early return if the staging table is empty; hard abort if `&syscc > 0`.
* `CLOSE_VARS` is not supported on REDSHIFT / POSTGRES / SNOWFLAKE engines (returns with a NOTE).
* A zero-row snapshot of the base table (`&basecopy`) is taken with `data ... set base; stop;` — this doubles as a lock check, since metadata functions fail against a locked table.
* `proc contents` on the base table feeds column lists. Columns are split into character and numeric lists for hashing (`%mp_rowhash` hashes the two types via separate arrays). Temporal columns, the processed column, and the delete-flag column are excluded from the hash.
* Table names containing `___TMP___` or `_____` are rejected (they would collide with generated temp columns).
### 2. Locking
For `LOADTARGET=YES`, the base table (and audit table, if configured) is locked via `%mp_lockanytable` before any staging prep, because the load is a two-part update (closeouts + append) and must not interleave with another load.
### 3. Staging preparation (`work.bitemp0_append`)
* If `RK_UNDERLYING` is configured, `%mp_retainedkey` maps business keys to retained keys (filtering the base lookup to live records, `&now < tech_to`, for temporal types); otherwise the staging table is used as-is.
* `bus_from` / `bus_to` overrides are applied if provided; the processed column is stamped with the load timestamp (`now`). Even with `processed=0`, a column literally named `PROCESSED_DTTM` on the base table is used if present.
* The MD5 change-detection hash is computed for every staged record.
* If the staging table contains the delete-flag column (`_____DELETE__THIS__RECORD_____`), rows flagged `"Yes"` are diverted to `&outds_del` and closed out via `%bitemporal_closeouts` (PK is taken as `bus_from` + business key). The remaining rows continue as `bitemp0_append`.
### 4. CLOSE_VARS closeout
When `CLOSE_VARS` (a subset of the PK) is supplied, live base records whose CLOSE_VARS values appear in staging but whose full PK does **not** are closed out. This handles "this group was fully reloaded, so anything missing was removed" semantics without reloading the whole table.
### 5. Uniqueness check
If `CHECK_UNIQUENESS=YES` (the default) or business-date overrides are in play, the staging table is sorted `nodupkey` by the PK; a row-count mismatch aborts the load and releases the locks. The staging table must be a unique snapshot of the business key at one point in business time.
### 6. Base extract (`work.bitemp0_base`)
Only base records matching staged PKs are extracted — for temporal load types, only **currently live** records (`now < tech_to`). A left join from the staged keys to the base produces `___TMP___NEW_FLG` to identify brand-new keys. This is engine-specific:
* **OLEDB (SQL Server)** — staged keys are pushed to a `##global` temp table and joined via explicit pass-through.
* **REDSHIFT / POSTGRES / SNOWFLAKE** — an in-database temp table is created `like` the base, stripped to PK + an added `md5 varchar(32)` column, loaded with staged keys, and joined via pass-through. Snowflake uses transient tables; Redshift gets `alter sortkey none` plus any `DCBL_REDSH` config options from `MPE_CONFIG`.
* **CAS** — a FedSQL join against a CASUSER copy.
* **BASE/other** — plain PROC SQL in SAS.
### 7. Change classification
* **`&outds_add`** — staged records flagged as new (no base match). They get `tech_from=now`, `tech_to=high_date`.
* **`work.bitemp1_current`** — matched base records, re-hashed so hashes are comparable.
* **Inserts (BITEMPORAL only)** — a staged record whose business range falls strictly *inside* an existing record's range splits the existing record into a "before" and "after" segment (`bitemp3_inserts` / `bitemp3a_inserts`). The split segments replace the original in the comparison set (`bitemp3b_newbase`).
* **Updates (`bitemp4*`)** — staged records matching a base PK but with a different hash or different business dates. Base and staged versions are stacked, deduplicated, then aligned in **two passes** over the business timeline per key: a forward pass (carry `bus_from` forward across identical hashes; a staged record trims the preceding base record's range) and a reverse pass (carry `bus_to` back; records fully subsumed by the new version are deleted). Records that end up byte-identical to what is already stored are dropped via a hash lookup that includes the business dates (`bitemp5a_lkp` / `bitemp5b_updates`, BITEMPORAL only).
### 8. Closeout application
Changed records are closed out in the target before the new versions are appended. SAS SQL has no UPDATE-with-join, so a correlated `EXISTS` subquery against `work.bitemp5d_subquery` is used (pushed in-database as a temp table for OLEDB / REDSHIFT / POSTGRES / SNOWFLAKE):
* **BITEMPORAL** — per key, the closeout range is `min(bus_from)` / `max(bus_to)` of the changed records: `update base set tech_to=now (, processed=now) where tech_from <= now < tech_to and exists (key match and base.bus_from >= min and base.bus_to <= max)`.
* **TXTEMPORAL** — same, on key only.
* **UPDATE** — changed records are physically `delete`d (they are re-appended in the next step). On CAS this uses `table.deleteRows` with a whereTable; temporal types are not supported on CAS and abort.
* **BUSTEMPORAL** — closeouts are not implemented; the macro aborts at this point ("BUSTEMPORAL NOT YET SUPPORTED").
### 9. Append, unlock, audit, log
* The union of modified + new records, deduplicated on all columns (`bitemp6_unique`), is appended to the base table (`proc append ... force nowarn`; CAS appends via a varchar-casting step; Redshift applies `DCBL_REDSH` options). Locks are then released.
* If `outds_audit` is set (always, from `%mpe_targetloader` — `AUDIT_LIBDS` defaulting to `&dclib..MPE_AUDIT`), `%mp_storediffs` compares the pre-load snapshot with the applied changes and appends row-level audit records; modified-row entries with no actual value change are removed (`MOVE_TYPE="M" and IS_PK=0 and IS_DIFF=0`).
* A summary row is inserted into `&dclib..MPE_DATALOADS` (libref, dsn, etlsource, loadtype, changed/new/deleted counts, duration, macro version, user, timestamp) unless `LOG=0`.
## REPLACE Load Type
REPLACE is the simplest load type and is deliberately **not** routed through `%bitemporal_dataloader` — it is implemented inline in `%mpe_targetloader` and simply wipes the target table (`delete * from`) and re-appends the staging table verbatim, with no history, change detection, key matching, audit rows, or `MPE_DATALOADS` logging. See [replace-load-type.md](replace-load-type.md) for the full deep dive.
## Supporting Tables
| Table | Role |
|---|---|
| `MPE_TABLES` | Per-table load configuration (loadtype, keys, temporal vars, audit target) |
| `MPE_SELECTBOX` | Dropdown values, including the LOADTYPE list (seeded in `mpe_makedata.sas`) |
| `MPE_SUBMIT` / `MPE_REVIEW` | Approval workflow state |
| `MPE_LOADS` | Submission-level log of CSV package loads |
| `MPE_DATALOADS` | Per-load statistics (temporal/UPDATE loads only) |
| `MPE_AUDIT` (or `AUDIT_LIBDS`) | Row-level change audit (temporal/UPDATE/FORMAT_CAT loads only) |
| `MPE_LOCKANYTABLE` | Lock control table |
| `MPE_CONFIG` | Engine-specific config (e.g. `DCBL_REDSH` scope for Redshift bulk options) |
## Testing
Unit tests for the loaders live next to the macros: `sas/sasjs/macros/bitemporal_dataloader.test.[1-4].sas` and `sas/sasjs/macros/mpe_targetloader.test.sas` (REPLACE loadtype), executed as sasjs tests. See [testing.md](testing.md) for how to run them (`npm run 4gl` then `sasjs test -t 4gl` from the `sas/` directory), and remember `sasjs lint` after touching any `.sas` files.
.agent/docs/replace-load-type.md
+97
View File
@@ -0,0 +1,97 @@
# REPLACE Load Type — Technical Deep Dive
This document describes the REPLACE load type in detail. For the overall loader architecture, the temporal model, and the other load types, see [bitemporal-dataloader.md](bitemporal-dataloader.md).
## Overview
REPLACE is the simplest load type and is deliberately **not** routed through `%bitemporal_dataloader`. It is implemented inline in `%mpe_targetloader` (`sas/sasjs/macros/mpe_targetloader.sas`, search for `&loadtype=REPLACE`). There is no history, no change detection, no key matching — the target table is wiped and reloaded from the staging table verbatim.
## Actual load (`LOADTARGET=YES`)
```sas
%mp_lockanytable(LOCK, lib=&lib, ds=&ds, ...)
data WORK.&STAGING_DS;
set WORK.&STAGING_DS;
/* only if the target contains the MPE_TABLES.VAR_PROCESSED variable: */
&VAR_PROCESSED = &now;
drop _____DELETE__THIS__RECORD_____;
run;
%if &engine_type=CAS %then %do;
/* prep first: cast varchar columns in a CASUSER copy of staging */
proc contents noprint data=&libds out=work.rpl_base_cols(keep=name type);
proc contents noprint data=WORK.&STAGING_DS out=work.rpl_stag_cols(keep=name type);
/* work.rpl_vchars = columns that are varchar in target AND in staging */
data casuser.&tmpds; /* temp copy with varchars casted */
length <varchar col> varchar(*); ...
set WORK.&STAGING_DS (rename=(<varchar col>=<tmp> ...));
<varchar col>=<tmp>; ... drop <tmp> ...;
run;
/* unlock + abort if any error so far - nothing destructive done yet */
%mp_abort(iftrue= (&syscc>0) ...)
/* destructive step, deliberately last before the append */
proc cas;
table.deleteRows / table={caslib="&lib",name="&ds",where="1=1"};
quit;
data &libds (append=yes) / sessref=dcsession;
set casuser.&tmpds;
run;
proc sql; drop table CASUSER.&tmpds; quit;
%end;
%else %do;
/* unlock + abort if any error so far - nothing destructive done yet */
%mp_abort(iftrue= (&syscc>0) ...)
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, ...)
```
Step by step:
1. **Lock** — the target is locked via `%mp_lockanytable` (control table `&dclib..MPE_LOCKANYTABLE`). This is the only concurrency guard for the whole operation.
2. **Staging recopy** — the staging dataset is rewritten in place: the processed-timestamp column is set to the approval timestamp **only if** the target table contains the variable named by `MPE_TABLES.VAR_PROCESSED` (unlike `%bitemporal_dataloader`, there is no `PROCESSED_DTTM` fallback). The delete-flag column `_____DELETE__THIS__RECORD_____` is unconditionally dropped — per-record delete semantics do not exist in REPLACE (everything is deleted anyway), so a delete flag submitted by the user is silently discarded (a `drop` of a non-existent variable just produces a warning).
3. **Prepare the load (CAS only)** — fixed char variables cannot be appended to CAS varchar columns, so the staging table is first copied to a CASUSER temp table with every column that is varchar in the target (and present in staging) redeclared as `varchar(*)` via generated `length` / `rename` / assignment statements. This mirrors the CAS append in `%bitemporal_dataloader`.
4. **Pre-destructive abort check** — if `&syscc > 0` after all preparation, the target is unlocked and the macro aborts via `%mp_abort`. Nothing destructive has happened at this point, so a failed prep leaves the target intact (and without a stale lock).
5. **Delete all rows** — every row is removed from the target while preserving structure, indexes and metadata, deliberately as the last step before the append (to minimise the time the target sits empty). On most engines this is `proc sql; delete * from &libds;` (passed through as a `DELETE` on database libraries). CAS tables do not support SQL deletes, so on the CAS engine the table is truncated instead via the `table.deleteRows` action with `where="1=1"` (same approach as `%bitemporal_closeouts`), with the libref passed as the caslib.
6. **Append staged rows** — the entire staging table is appended. On most engines this is `proc append ... force nowarn` (`force` allows the append to proceed despite attribute mismatches, so lengths/formats may be coerced or values truncated, and columns present in only one side are handled; `nowarn` suppresses the associated warnings). On CAS the pre-cast CASUSER temp table is appended with `data &libds (append=yes) / sessref=dcsession` and then dropped.
7. **Unlock**.
## Diff screen (`LOADTARGET=NO`)
When the approver reviews a REPLACE submission, no real comparison is performed:
```sas
/* 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; /* every staged record is "NEW" */
run; /* outds_mod stays empty */
/* previous table will be considered fully deleted */
data work.outds_del;
set &lib..&ds; /* every existing record is "DELETED" */
run;
```
The approval screen therefore always shows full-table turnover: the entire current table as deleted and the entire staging table as added, with no "modified" records and no original/current comparison.
## What REPLACE does not do
Because it bypasses `%bitemporal_dataloader`, none of the following apply to REPLACE:
* **No row-level audit** — nothing is written to `AUDIT_LIBDS` / `MPE_AUDIT`. The record of the change is the approval package itself (staged CSV and the `TEMPDIFFS` CSV stored under `&mpelocapprovals/&LOAD_REF`).
* **No `MPE_DATALOADS` log entry** — the load-log insert lives inside `%bitemporal_dataloader`, so REPLACE loads do not appear in load history. (Consequently the SHOW_DIFFS timestamp lookup in `postdata.sas` finds no `MPE_DATALOADS` row for a REPLACE load and falls back to the current datetime.)
* **No PK usage** — `BUSKEY` is ignored: no uniqueness check, no dedup, no join to existing data. Staged rows are loaded exactly as submitted, duplicates included.
* **No temporal handling** — `VAR_TXFROM`/`VAR_TXTO`/`VAR_BUSFROM`/`VAR_BUSTO` and `CLOSE_VARS` are ignored. A REPLACE table should not be treated as temporal; querying it with the usual validity filters makes no sense.
* **No retained-key handling** — `RK_UNDERLYING` is ignored.
* **Minimal engine-specific handling** — unlike the temporal loaders, there is no pass-through/temp-table optimisation for OLEDB, Redshift, Postgres or Snowflake. The only engine conditional is CAS, where rows are removed via the `deleteRows` action and appended via a varchar-casting CASUSER temp table and a data-step append (SQL deletes and fixed-char-to-varchar appends are not possible on CAS).
* **Not atomic** — the delete/truncate and append are separate steps with no transaction or rollback, so a session failure between them still leaves the target empty or partially loaded (recovery is a re-approval of the same or a previous staging package). The risk is mitigated by performing all preparation first, aborting on any error before the destructive step, and executing the delete/truncate immediately before the append.
## Interaction with Row Level Security
REPLACE is incompatible with `EDIT`-scope RLS rules (a full-table wipe cannot honour row-level write restrictions); this is enforced at edit time in both directions — see [row-level-security.md](row-level-security.md#incompatibility-with-replace-load-type) and [issue #211](https://git.datacontroller.io/dc/dc/issues/211). `VIEW`-scope rules remain compatible.
.agent/skills/sas/SKILL.md
+73
View File
@@ -0,0 +1,73 @@
---
name: sas
description: >
Use this skill whenever writing, modifying, or debugging SAS code in this repository — SAS
macros, sasjs services, hooks, or test files (*.sas). Triggers include: SAS macro compilation
errors, parameter parsing issues, writing sasjs tests with mp_assert / mp_assertdsobs /
mp_assertscope, macro-quoting questions (%str, %nrstr, %superq), %mp_abort usage, or unexpected
behaviour when passing free-text values (descriptions, messages) to macros. Also covers repo
conventions such as running `sasjs lint` after touching .sas files and the maximum line length
rule.
---
# SAS Development
Conventions and pitfalls for SAS code in this repository (macros in `sas/sasjs/macros/`, services
in `sas/sasjs/services/`, tests alongside them as `*.test.sas`).
## Always `%str()` free-text macro parameters
A comma inside a macro parameter value is parsed as a **parameter delimiter**. Free-text
parameters such as `desc=` (in `%mp_assert`, `%mp_assertdsobs`, `%mp_assertscope`) and `msg=` (in
`%mp_abort`) must be wrapped in `%str()` — even when they currently contain no comma, so later
edits cannot silently break the call.
Wrong — the text after the comma becomes an unexpected positional parameter and the compilation
fails (or worse, is misparsed):
```sas
%mp_assert(iftrue=(&oldrows=0),
desc=Test 2 - all staged records loaded, delete flags ignored,
outds=work.test_results
)
```
Right:
```sas
%mp_assert(iftrue=(&oldrows=0),
desc=%str(Test 2 - all staged records loaded, delete flags ignored),
outds=work.test_results
)
```
The same applies to any macro parameter that carries user-facing text: `%mp_abort(msg=%str(...))`,
`etlsource=` values containing punctuation (use `%superq()` when passing macro variables that may
contain special characters), etc.
Related quoting rules of thumb:
- `%str()` masks commas, parentheses, semicolons and quotes at compile time — sufficient for
static text like `desc=`.
- Use `%nrstr()` if the text must also mask `%` and `&` (rare in descriptions).
- Use `%superq(var)` when forwarding a macro **variable** whose value may contain special
characters (e.g. `etlsource=` in `mpe_targetloader.sas`).
## Tests must be idempotent
A test file must pass when run repeatedly (including after a run that failed partway). Conventions:
- Start the file with `%let syscc=0;` — many macros (e.g. `%mp_lockanytable(LOCK)`) abort on entry if `&syscc>0`, and any `WARNING` in a previous test bumps `syscc` to 4.
- Clean up **all** persistent state, not just the obvious one: `MPE_TABLES` registrations, `MPE_LOCKANYTABLE` lock records (a run dying between LOCK and UNLOCK leaves a stale `LOCKED` row), physical tables in `dctest` (`proc datasets ... delete`), and global macro variables created via `select ... into:` (`%symdel`).
- Make prep defensive: delete-then-insert config records (handles leftovers from an aborted run), and recreate physical tables rather than assuming they are absent.
## Repo conventions for .sas files
- Run `npx sasjs lint` from the `sas/` directory after creating or modifying any `.sas` file; the
files you touched must have zero warnings (pre-existing warnings in other files can be ignored).
- Maximum line length is 80 characters (lint-enforced) — this is why long `msg=%str(...)` values
sometimes need shortening.
- Test files are named `<thing>.test.sas` (or `<thing>.test.N.sas`) and run via
`npm run 4gl && sasjs test -t 4gl` from `sas/` — see `.agent/docs/testing.md`.
- `sas/sasjsbuild/` is generated build output — never hand-edit it; edit sources under
`sas/sasjs/` only.
sas/sasjs/macros/mpe_targetloader.sas
+96 -2
View File
@@ -12,6 +12,8 @@
<h4> SAS Macros </h4>
@li bitemporal_dataloader.sas
@li mf_existvar.sas
@li mf_getengine.sas
@li mf_getuniquename.sas
@li mp_abort.sas
@li mp_loadformat.sas
@li mp_lockanytable.sas
@@ -136,8 +138,100 @@ run;
%end;
drop _____DELETE__THIS__RECORD_____;
run;
proc sql; delete * from &libds;
proc append base=&libds data=WORK.&STAGING_DS force nowarn;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
)
%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;
sas/sasjs/macros/mpe_targetloader.test.sas
+178
View File
@@ -0,0 +1,178 @@
/**
@file
@brief Testing mpe_targetloader macro - REPLACE loadtype
@details Covers the REPLACE branch of mpe_targetloader.sas:
* LOADTARGET=NO (diff screen preparation) - all staged records classed as
new, all existing records classed as deleted, target table unchanged
* LOADTARGET=YES (actual load) - target table fully replaced by the
staging table, delete flag column dropped, processed column stamped,
per-record delete flags ignored (everything is loaded)
A dedicated DCTEST.DC_REPLACE table is registered in MPE_TABLES (and the
registration removed again in cleanup). The DCTEST library is BASE engine,
so the CAS-specific branch (deleteRows truncation / varchar casting) is not
exercised here.
<h4> SAS Macros </h4>
@li mp_assert.sas
@li mp_assertdsobs.sas
@li mp_assertscope.sas
@li mpe_targetloader.sas
@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.
**/
%let syscc=0;
/**
* Prep - physical target table and MPE_TABLES registration
*/
data dctest.dc_replace;
length pk $8 val $20;
pk='OLD1'; val='oldvalue1'; processed_dttm=0; output;
pk='OLD2'; val='oldvalue2'; processed_dttm=0; output;
format processed_dttm datetime19.;
run;
proc sql noprint;
delete from &dc_libref..mpe_tables where libref="DCTEST" and dsn='DC_REPLACE';
insert into &dc_libref..mpe_tables
set tx_from=0
,tx_to='31DEC5999:23:59:59'dt
,libref="DCTEST"
,dsn='DC_REPLACE'
,buskey='PK'
,loadtype='REPLACE'
,var_processed='PROCESSED_DTTM'
,num_of_approvals_required=1;
quit;
/* staging table, as it would arrive from the approval package */
data work.staging_ds;
length pk $8 val $20 _____DELETE__THIS__RECORD_____ $3;
pk='NEW1'; val='newvalue1'; _____DELETE__THIS__RECORD_____='No'; output;
pk='NEW2'; val='newvalue2'; _____DELETE__THIS__RECORD_____='Yes'; output;
pk='NEW3'; val='newvalue3'; _____DELETE__THIS__RECORD_____='No'; output;
run;
/**
* Test 1 - LOADTARGET=NO builds the diff tables without touching the target
*/
%mp_assertscope(SNAPSHOT)
%mpe_targetloader(libds=DCTEST.DC_REPLACE
,etlsource=mpe_targetloader.test
,STAGING_DS=STAGING_DS
,LOADTARGET=NO
,dclib=&dc_libref
,dc_dttmtfmt=&dc_dttmtfmt.
)
%mp_assertscope(COMPARE,
desc=%str(Test 1 - checking macro variables against previous snapshot)
)
%mp_assert(iftrue=(&syscc=0),
desc=%str(Test 1 - REPLACE LOADTARGET=NO completed without errors),
outds=work.test_results
)
%mp_assertdsobs(work.outds_add,
desc=%str(Test 1 - all staged records classed as new),
test=EQUALS 3,
outds=work.test_results
)
%mp_assertdsobs(work.outds_del,
desc=%str(Test 1 - all existing records classed as deleted),
test=EQUALS 2,
outds=work.test_results
)
%mp_assertdsobs(work.outds_mod,
desc=%str(Test 1 - no records classed as modified),
test=EQUALS 0,
outds=work.test_results
)
%mp_assertdsobs(dctest.dc_replace,
desc=%str(Test 1 - target table not modified),
test=EQUALS 2,
outds=work.test_results
)
/**
* Test 2 - LOADTARGET=YES replaces the target table with the staged data
*/
%mp_assertscope(SNAPSHOT)
%mpe_targetloader(libds=DCTEST.DC_REPLACE
,etlsource=mpe_targetloader.test
,STAGING_DS=STAGING_DS
,LOADTARGET=YES
,dclib=&dc_libref
,dc_dttmtfmt=&dc_dttmtfmt.
)
%mp_assertscope(COMPARE,
desc=%str(Test 2 - checking macro variables against previous snapshot)
)
%mp_assert(iftrue=(&syscc=0),
desc=%str(Test 2 - REPLACE LOADTARGET=YES completed without errors),
outds=work.test_results
)
%mp_assertdsobs(dctest.dc_replace,
desc=%str(Test 2 - target row count matches staging table),
test=EQUALS 3,
outds=work.test_results
)
proc sql noprint;
select count(*) into: oldrows
from dctest.dc_replace where pk in ('OLD1','OLD2');
select count(*) into: newrows
from dctest.dc_replace where pk in ('NEW1','NEW2','NEW3');
select count(*) into: delcol from dictionary.columns
where libname='DCTEST' and memname='DC_REPLACE'
and upcase(name)='_____DELETE__THIS__RECORD_____';
select count(*) into: notstamped
from dctest.dc_replace where missing(processed_dttm) or processed_dttm=0;
quit;
%mp_assert(iftrue=(&oldrows=0),
desc=%str(Test 2 - pre-existing records removed from target),
outds=work.test_results
)
%mp_assert(iftrue=(&newrows=3),
desc=%str(Test 2 - all staged records loaded, delete flags ignored),
outds=work.test_results
)
%mp_assert(iftrue=(&delcol=0),
desc=%str(Test 2 - delete flag column not loaded to target),
outds=work.test_results
)
%mp_assert(iftrue=(&notstamped=0),
desc=%str(Test 2 - processed_dttm stamped on every loaded record),
outds=work.test_results
)
/**
* Cleanup - remove all persistent state created by this test, so that a
* subsequent run starts from the same position (including a run that
* previously failed partway through)
*/
proc sql noprint;
/* REPLACE table registration */
delete from &dc_libref..mpe_tables where libref="DCTEST" and dsn='DC_REPLACE';
/* lock record (may be left as LOCKED after an aborted run) */
delete from &dc_libref..mpe_lockanytable
where lock_lib="DCTEST" and lock_ds="DC_REPLACE";
quit;
/* physical target table */
proc datasets lib=dctest nolist;
delete dc_replace;
run;
quit;
/* assertion macro variables */
%symdel oldrows newrows delcol notstamped;