From 1f62b552636a22376f766e0a00775f5408ad3404 Mon Sep 17 00:00:00 2001
From: Victor San Kho Lin <victor@sankholin.com>
Date: Sun, 12 Jan 2025 00:16:50 +1100
Subject: [PATCH] Implemented OrcaVault psa schema

* Persistent Staging Area (psa) is a layer in data warehouse where data is
  kept archived and track change history. The data table in this area also
  act as an intermediate storage location to provide downstream transformation.
* Made use of spreadsheet_library_tracking_metadata table as another data
  source to feed into vault layer for change history and data consolidation use.
* Typically, psa data tables is processed by dbt with append only incremental
  materialization and the data is sourced from tsa schema table counterpart.
* Added `next.sh` script and `tsa.truncate_tables()` db function to simulate
  incremental data loading between tsa to psa transformation in local dev setup.
---
 dev/Makefile                                  |  15 ++-
 dev/src/next.sh                               |   7 +
 dev/src/psa.sql                               |  38 ++++++
 dev/src/tsa.sql                               |  23 ++++
 orcavault/Makefile                            |   9 +-
 orcavault/dbt_project.yml                     |   3 +
 orcavault/models/psa/schema.yml               |  60 +++++++++
 .../spreadsheet_library_tracking_metadata.sql | 123 ++++++++++++++++++
 8 files changed, 276 insertions(+), 2 deletions(-)
 create mode 100644 dev/src/next.sh
 create mode 100644 dev/src/psa.sql
 create mode 100644 orcavault/models/psa/schema.yml
 create mode 100644 orcavault/models/psa/spreadsheet_library_tracking_metadata.sql

diff --git a/dev/Makefile b/dev/Makefile
index d38b67a..e4b214f 100644
--- a/dev/Makefile
+++ b/dev/Makefile
@@ -31,11 +31,24 @@ tsa:
 	@docker compose cp ./src/tsa.sql postgres:/tmp/tsa.sql
 	@docker compose exec -e PGPASSWORD=dev -it postgres psql -h 0.0.0.0 -d orcavault -U dev -f /tmp/tsa.sql
 
-all: ods tsa
+psa:
+	@docker compose cp ./src/psa.sql postgres:/tmp/psa.sql
+	@docker compose exec -e PGPASSWORD=dev -it postgres psql -h 0.0.0.0 -d orcavault -U dev -f /tmp/psa.sql
+
+all: ods tsa psa
 
 load:
 	@docker compose cp ./src/load.sh postgres:/tmp/load.sh
 	@docker compose exec -it postgres bash /tmp/load.sh
 
+wait:
+	@sleep 5
+
+reload: down up wait all load
+
+next:
+	@docker compose cp ./src/next.sh postgres:/tmp/next.sh
+	@docker compose exec -it postgres bash /tmp/next.sh
+
 sync:
 	@aws s3 sync s3://orcabus-test-data-843407916570-ap-southeast-2/orcavault/data/ data/
diff --git a/dev/src/next.sh b/dev/src/next.sh
new file mode 100644
index 0000000..0987968
--- /dev/null
+++ b/dev/src/next.sh
@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+
+PGPASSWORD=dev psql -h 0.0.0.0 -d orcavault -U dev -c 'SELECT tsa.truncate_tables();'
+
+PGPASSWORD=dev psql -h 0.0.0.0 -d orcavault -U dev <<EOF
+\copy tsa.spreadsheet_library_tracking_metadata from '/data/orcavault_tsa_spreadsheet_library_tracking_metadata.next.csv' with (format csv, header true, delimiter ',');
+EOF
diff --git a/dev/src/psa.sql b/dev/src/psa.sql
new file mode 100644
index 0000000..e2c5ae8
--- /dev/null
+++ b/dev/src/psa.sql
@@ -0,0 +1,38 @@
+SELECT current_database();
+
+-- create psa schema
+CREATE SCHEMA IF NOT EXISTS psa AUTHORIZATION dev;
+SET search_path TO psa;
+
+SELECT current_schema();
+
+CREATE TABLE IF NOT EXISTS orcavault.psa.spreadsheet_library_tracking_metadata
+(
+    assay                 varchar,
+    comments              varchar,
+    coverage              varchar,
+    experiment_id         varchar,
+    external_sample_id    varchar,
+    external_subject_id   varchar,
+    library_id            varchar,
+    override_cycles       varchar,
+    phenotype             varchar,
+    project_name          varchar,
+    project_owner         varchar,
+    qpcr_id               varchar,
+    quality               varchar,
+    run                   varchar,
+    sample_id             varchar,
+    sample_name           varchar,
+    samplesheet_sample_id varchar,
+    source                varchar,
+    subject_id            varchar,
+    truseq_index          varchar,
+    type                  varchar,
+    workflow              varchar,
+    r_rna                 varchar,
+    study                 varchar,
+    sheet_name            varchar,
+    load_datetime         timestamptz,
+    record_source         varchar(255)
+);
diff --git a/dev/src/tsa.sql b/dev/src/tsa.sql
index 3134f1a..328ae2c 100644
--- a/dev/src/tsa.sql
+++ b/dev/src/tsa.sql
@@ -6,6 +6,29 @@ SET search_path TO tsa;
 
 SELECT current_schema();
 
+-- --
+
+DROP FUNCTION IF EXISTS tsa.truncate_tables();
+
+CREATE OR REPLACE FUNCTION tsa.truncate_tables()
+    RETURNS void
+    LANGUAGE 'sql'
+    COST 100
+    VOLATILE PARALLEL UNSAFE
+AS $BODY$
+DO $$ DECLARE
+    table_name text;
+BEGIN
+    FOR table_name IN (SELECT tablename FROM pg_tables WHERE schemaname='tsa') LOOP
+        EXECUTE 'TRUNCATE TABLE tsa."' || table_name || '" CASCADE;';
+    END LOOP;
+END $$;
+$BODY$;
+
+ALTER FUNCTION tsa.truncate_tables() OWNER TO dev;
+
+-- --
+
 CREATE TABLE IF NOT EXISTS orcavault.tsa.spreadsheet_library_tracking_metadata
 (
     assay                 varchar,
diff --git a/orcavault/Makefile b/orcavault/Makefile
index 92ffce4..43a940a 100644
--- a/orcavault/Makefile
+++ b/orcavault/Makefile
@@ -26,13 +26,20 @@ ods:
 tsa:
 	@(cd ../dev/ && $(MAKE) tsa)
 
+psa:
+	@(cd ../dev/ && $(MAKE) psa)
+
 all:
 	@(cd ../dev/ && $(MAKE) all)
 
 load:
 	@(cd ../dev/ && $(MAKE) load)
 
-reload: down up ods tsa load
+reload:
+	@(cd ../dev/ && $(MAKE) reload)
+
+next:
+	@(cd ../dev/ && $(MAKE) next)
 
 sync:
 	@(cd ../dev/ && $(MAKE) sync)
diff --git a/orcavault/dbt_project.yml b/orcavault/dbt_project.yml
index 5d4a0ec..4fe697c 100644
--- a/orcavault/dbt_project.yml
+++ b/orcavault/dbt_project.yml
@@ -18,6 +18,9 @@ clean-targets:
 models:
   orcavault:
     +materialized: view
+    psa:
+      +schema: psa
+      +materialized: table
     raw:
       +schema: raw
       +materialized: table
diff --git a/orcavault/models/psa/schema.yml b/orcavault/models/psa/schema.yml
new file mode 100644
index 0000000..c312515
--- /dev/null
+++ b/orcavault/models/psa/schema.yml
@@ -0,0 +1,60 @@
+version: 2
+
+models:
+
+  - name: spreadsheet_library_tracking_metadata
+    columns:
+      - name: assay
+        data_type: varchar
+      - name: comments
+        data_type: varchar
+      - name: coverage
+        data_type: varchar
+      - name: experiment_id
+        data_type: varchar
+      - name: external_sample_id
+        data_type: varchar
+      - name: external_subject_id
+        data_type: varchar
+      - name: library_id
+        data_type: varchar
+      - name: override_cycles
+        data_type: varchar
+      - name: phenotype
+        data_type: varchar
+      - name: project_name
+        data_type: varchar
+      - name: project_owner
+        data_type: varchar
+      - name: qpcr_id
+        data_type: varchar
+      - name: quality
+        data_type: varchar
+      - name: run
+        data_type: varchar
+      - name: sample_id
+        data_type: varchar
+      - name: sample_name
+        data_type: varchar
+      - name: samplesheet_sample_id
+        data_type: varchar
+      - name: source
+        data_type: varchar
+      - name: subject_id
+        data_type: varchar
+      - name: truseq_index
+        data_type: varchar
+      - name: type
+        data_type: varchar
+      - name: workflow
+        data_type: varchar
+      - name: r_rna
+        data_type: varchar
+      - name: study
+        data_type: varchar
+      - name: sheet_name
+        data_type: varchar
+      - name: load_datetime
+        data_type: timestamptz
+      - name: record_source
+        data_type: varchar(255)
diff --git a/orcavault/models/psa/spreadsheet_library_tracking_metadata.sql b/orcavault/models/psa/spreadsheet_library_tracking_metadata.sql
new file mode 100644
index 0000000..f025dda
--- /dev/null
+++ b/orcavault/models/psa/spreadsheet_library_tracking_metadata.sql
@@ -0,0 +1,123 @@
+{{
+    config(
+        materialized='incremental'
+    )
+}}
+
+with source as (
+
+    select * from {{ source('tsa', 'spreadsheet_library_tracking_metadata') }}
+
+),
+
+cleaned as (
+
+    select
+        trim(regexp_replace(assay, E'[\\n\\r]+', '', 'g')) as assay,
+        trim(regexp_replace(comments, E'[\\n\\r]+', '', 'g')) as comments,
+        trim(regexp_replace(coverage, E'[\\n\\r]+', '', 'g')) as coverage,
+        trim(regexp_replace(experiment_id, E'[\\n\\r]+', '', 'g')) as experiment_id,
+        trim(regexp_replace(external_sample_id, E'[\\n\\r]+', '', 'g')) as external_sample_id,
+        trim(regexp_replace(external_subject_id, E'[\\n\\r]+', '', 'g')) as external_subject_id,
+        trim(regexp_replace(library_id, E'[\\n\\r]+', '', 'g')) as library_id,
+        trim(regexp_replace(override_cycles, E'[\\n\\r]+', '', 'g')) as override_cycles,
+        trim(regexp_replace(phenotype, E'[\\n\\r]+', '', 'g')) as phenotype,
+        trim(regexp_replace(project_name, E'[\\n\\r]+', '', 'g')) as project_name,
+        trim(regexp_replace(project_owner, E'[\\n\\r]+', '', 'g')) as project_owner,
+        trim(regexp_replace(qpcr_id, E'[\\n\\r]+', '', 'g')) as qpcr_id,
+        trim(regexp_replace(quality, E'[\\n\\r]+', '', 'g')) as quality,
+        trim(regexp_replace(run, E'[\\n\\r]+', '', 'g')) as run,
+        trim(regexp_replace(sample_id, E'[\\n\\r]+', '', 'g')) as sample_id,
+        trim(regexp_replace(sample_name, E'[\\n\\r]+', '', 'g')) as sample_name,
+        trim(regexp_replace(samplesheet_sample_id, E'[\\n\\r]+', '', 'g')) as samplesheet_sample_id,
+        trim(regexp_replace(source, E'[\\n\\r]+', '', 'g')) as source,
+        trim(regexp_replace(subject_id, E'[\\n\\r]+', '', 'g')) as subject_id,
+        trim(regexp_replace(truseq_index, E'[\\n\\r]+', '', 'g')) as truseq_index,
+        trim(regexp_replace(type, E'[\\n\\r]+', '', 'g')) as type,
+        trim(regexp_replace(workflow, E'[\\n\\r]+', '', 'g')) as workflow,
+        trim(regexp_replace(r_rna, E'[\\n\\r]+', '', 'g')) as r_rna,
+        trim(regexp_replace(study, E'[\\n\\r]+', '', 'g')) as study,
+        trim(regexp_replace(sheet_name, E'[\\n\\r]+', '', 'g')) as sheet_name
+    from
+        source
+    where
+        coalesce
+        (
+            nullif(assay, ''),
+            nullif(comments, ''),
+            nullif(coverage, ''),
+            nullif(experiment_id, ''),
+            nullif(external_sample_id, ''),
+            nullif(external_subject_id, ''),
+            nullif(library_id, ''),
+            nullif(override_cycles, ''),
+            nullif(phenotype, ''),
+            nullif(project_name, ''),
+            nullif(project_owner, ''),
+            nullif(qpcr_id, ''),
+            nullif(quality, ''),
+            nullif(run, ''),
+            nullif(sample_id, ''),
+            nullif(sample_name, ''),
+            nullif(samplesheet_sample_id, ''),
+            nullif(source, ''),
+            nullif(subject_id, ''),
+            nullif(truseq_index, ''),
+            nullif(type, ''),
+            nullif(workflow, ''),
+            nullif(r_rna, ''),
+            nullif(study, '')
+        ) is not null
+
+),
+
+transformed as (
+
+    select
+        *
+    from
+        cleaned
+    except
+    select
+        assay,
+        comments,
+        coverage,
+        experiment_id,
+        external_sample_id,
+        external_subject_id,
+        library_id,
+        override_cycles,
+        phenotype,
+        project_name,
+        project_owner,
+        qpcr_id,
+        quality,
+        run,
+        sample_id,
+        sample_name,
+        samplesheet_sample_id,
+        source,
+        subject_id,
+        truseq_index,
+        type,
+        workflow,
+        r_rna,
+        study,
+        sheet_name
+    from
+        {{ this }}
+
+),
+
+final as (
+
+    select
+        *,
+        cast('{{ run_started_at }}' as timestamptz) as load_datetime,
+        (select 'UMCCR_Library_Tracking_MetaData') as record_source
+    from
+        transformed
+
+)
+
+select * from final