## ----include = FALSE----------------------------------------------------------
# Skip this vignette when the optional packages are missing, or when the
# ducklake DuckDB extension is not already installed. The probe never
# downloads anything, so building this vignette needs no network access.
has_deps <- requireNamespace("pharmaversesdtm", quietly = TRUE) &&
  requireNamespace("admiral", quietly = TRUE)
ducklake_available <- ducklake::ducklake_extension_available()
knitr::opts_chunk$set(
  collapse = TRUE,
  comment = "#>",
  eval = has_deps && ducklake_available
)
# Use a unique temp directory for this vignette to avoid conflicts during R CMD check
vignette_temp_dir <- file.path(tempdir(), "clinical_trial_vignette")
dir.create(vignette_temp_dir, showWarnings = FALSE, recursive = TRUE)
knitr::opts_knit$set(root.dir = vignette_temp_dir)

## ----setup, message=FALSE-----------------------------------------------------
library(ducklake)
library(dplyr)
library(tidyr)
library(pharmaversesdtm)
library(admiral)
library(lubridate)
library(stringr)

## ----create-datalake----------------------------------------------------------
# Install the ducklake extension to duckdb (required once per system)
# The ducklake extension only needs installing once per machine:
# install_ducklake()

# Define where to create a new data lake or access an existing one
# For this vignette, we use vignette_temp_dir; in practice, use a permanent location
# lake_path <- "/path/to/your/project/data_lake"
trial_lake_path <- vignette_temp_dir

# attach_ducklake creates or attaches (if it already exists) a data lake
attach_ducklake(
  ducklake_name = "clinical_trial_lake",
  lake_path = trial_lake_path
)

# Verify the lake was created
list.files(trial_lake_path, pattern = "clinical_trial_lake")

## ----architecture-diagram, echo=FALSE, message=FALSE, warning=FALSE-----------
if (requireNamespace("DiagrammeR", quietly = TRUE)) {
  DiagrammeR::mermaid("
  graph LR
    A[Bronze Layer<br/>Raw Data<br/>dm_raw, ae_raw, ...] --> B[Silver Layer<br/>Cleaned Data<br/>dm, ae, ...]
    B --> C[Gold Layer<br/>Analysis Data<br/>adsl, adae, ...]
    
    style A fill:#cd7f32,stroke:#8b4513,stroke-width:2px,color:#fff
    style B fill:#c0c0c0,stroke:#808080,stroke-width:2px,color:#000
    style C fill:#ffd700,stroke:#daa520,stroke-width:2px,color:#000
  ", width = "100%", height = 200)
}

## ----load-dm------------------------------------------------------------------
# Bronze layer: Load raw SDTM Demographics exactly as received
with_transaction(
  create_table(pharmaversesdtm::dm, "dm_raw"),
  author = "T Gerke",
  commit_message = "Add raw demographics"
)

# Silver layer: Apply cleaning transformations
with_transaction(
  get_ducklake_table("dm_raw") |> 
    admiral::convert_blanks_to_na() |> 
    create_table("dm"),
  author = "T Gerke",
  commit_message = "Clean demographics data"
)

# Verify the cleaned table
get_ducklake_table("dm") |>
  select(USUBJID, AGE, SEX, RACE, ARM) |>
  head()

## ----dm-labels----------------------------------------------------------------
get_table_comments("dm_raw") |> head()

## ----dm-labels-collect--------------------------------------------------------
dm_check <- get_ducklake_table("dm_raw") |>
  select(USUBJID, AGE, SEX) |>
  collect()

attr(dm_check$AGE, "label")

## ----load-suppdm--------------------------------------------------------------
# Bronze layer: Raw data
with_transaction(
  create_table(pharmaversesdtm::suppdm, "suppdm_raw"),
  author = "T Gerke",
  commit_message = "Add raw supplemental demographics"
)

# Silver layer: Cleaned data
with_transaction(
  get_ducklake_table("suppdm_raw") |> 
    admiral::convert_blanks_to_na() |> 
    create_table("suppdm"),
  author = "T Gerke",
  commit_message = "Clean supplemental demographics"
)

## ----load-ds------------------------------------------------------------------
# Bronze layer
with_transaction(
  create_table(pharmaversesdtm::ds, "ds_raw"),
  author = "T Gerke",
  commit_message = "Add raw disposition"
)

# Silver layer
with_transaction(
  ds <- get_ducklake_table("ds_raw") |> 
    admiral::convert_blanks_to_na() |> 
    create_table("ds"),
  author = "T Gerke",
  commit_message = "Clean disposition data"
)

## ----load-ex------------------------------------------------------------------
# Bronze layer
with_transaction(
  create_table(pharmaversesdtm::ex, "ex_raw"),
  author = "T Gerke",
  commit_message = "Add raw exposure"
)

# Silver layer
with_transaction(
  ex <- get_ducklake_table("ex_raw") |> 
    admiral::convert_blanks_to_na() |> 
    create_table("ex"),
  author = "T Gerke",
  commit_message = "Clean exposure data"
)

## ----load-ae------------------------------------------------------------------
# Bronze layer
with_transaction(
  create_table(pharmaversesdtm::ae, "ae_raw"),
  author = "T Gerke",
  commit_message = "Add raw adverse events"
)

# Silver layer
with_transaction(
  ae <- get_ducklake_table("ae_raw") |> 
    admiral::convert_blanks_to_na() |> 
    create_table("ae"),
  author = "T Gerke",
  commit_message = "Clean adverse events"
)

## ----load-vs------------------------------------------------------------------
# Bronze layer
with_transaction(
  create_table(pharmaversesdtm::vs, "vs_raw"),
  author = "T Gerke",
  commit_message = "Add raw vital signs"
)

# Silver layer
with_transaction(
  vs <- get_ducklake_table("vs_raw") |> 
    admiral::convert_blanks_to_na() |> 
    create_table("vs"),
  author = "T Gerke",
  commit_message = "Clean vital signs"
)

## ----load-pc------------------------------------------------------------------
# Bronze layer
with_transaction(
  create_table(pharmaversesdtm::pc, "pc_raw"),
  author = "T Gerke",
  commit_message = "Add raw PK concentrations"
)

# Silver layer
with_transaction(
  pc <- get_ducklake_table("pc_raw") |> 
    admiral::convert_blanks_to_na() |> 
    create_table("pc"),
  author = "T Gerke",
  commit_message = "Clean PK concentrations"
)

## ----verify-sdtm-versions-----------------------------------------------------
# View the first 5 of all snapshots in the data lake
list_table_snapshots() |>
  head(5)

# Filter snapshots for specific tables
list_table_snapshots("dm_raw")
list_table_snapshots("dm")

## ----build-adsl---------------------------------------------------------------
# Read SDTM data from the lake and collect into memory
# Admiral functions require tibbles/data.frames, not lazy database connections
dm <- get_ducklake_table("dm") |> collect()
suppdm <- get_ducklake_table("suppdm") |> collect()
ds <- get_ducklake_table("ds") |> collect()
ex <- get_ducklake_table("ex") |> collect()
ae <- get_ducklake_table("ae") |> collect()
vs <- get_ducklake_table("vs") |> collect()

# Combine DM and SUPPDM
dm_suppdm <- dm |> 
  left_join(
    suppdm |> 
      filter(QNAM %in% c("EDUCLVL", "DISCONFL", "DSRAEFL")) |> 
      pivot_wider(
        id_cols = c(STUDYID, USUBJID),
        names_from = QNAM,
        values_from = QVAL
      ),
    by = c("STUDYID", "USUBJID")
  )

# Derive treatment dates and durations
ex_ext <- ex |> 
  derive_vars_dtm(
    dtc = EXSTDTC,
    new_vars_prefix = "EXST"
  ) |> 
  derive_vars_dtm(
    dtc = EXENDTC,
    new_vars_prefix = "EXEN",
    time_imputation = "last"
  )

# Derive disposition variables first (needed for later derivations)
ds_ext <- ds |> 
  derive_vars_dt(
    dtc = DSSTDTC,
    new_vars_prefix = "DSST"
  )

# Build ADSL with all derivations in a single pipeline
adsl <- dm_suppdm |> 
  # Treatment Start Datetime
  derive_vars_merged(
    dataset_add = ex_ext,
    filter_add = (EXDOSE > 0 | (EXDOSE == 0 & str_detect(EXTRT, "PLACEBO"))) & 
                 !is.na(EXSTDTM),
    new_vars = exprs(TRTSDTM = EXSTDTM),
    order = exprs(EXSTDTM, EXSEQ),
    mode = "first",
    by_vars = exprs(STUDYID, USUBJID)
  ) |> 
  # Treatment End Datetime
  derive_vars_merged(
    dataset_add = ex_ext,
    filter_add = (EXDOSE > 0 | (EXDOSE == 0 & str_detect(EXTRT, "PLACEBO"))) & 
                 !is.na(EXENDTM),
    new_vars = exprs(TRTEDTM = EXENDTM),
    order = exprs(EXENDTM, EXSEQ),
    mode = "last",
    by_vars = exprs(STUDYID, USUBJID)
  ) |> 
  # Convert to dates
  derive_vars_dtm_to_dt(source_vars = exprs(TRTSDTM, TRTEDTM)) |> 
  # Treatment duration
  derive_var_trtdurd() |> 
  # Safety population flag
  derive_var_merged_exist_flag(
    dataset_add = ex,
    by_vars = exprs(STUDYID, USUBJID),
    new_var = SAFFL,
    condition = (EXDOSE > 0 | (EXDOSE == 0 & str_detect(EXTRT, "PLACEBO")))
  ) |> 
  # Treatment variables
  mutate(
    TRT01P = ARM,
    TRT01A = ACTARM
  ) |> 
  # Age groups
  mutate(
    AGEGR1 = case_when(
      AGE < 18 ~ "<18",
      between(AGE, 18, 64) ~ "18-64",
      AGE > 64 ~ ">64",
      TRUE ~ "Missing"
    ),
    AGEGR1N = case_when(
      AGE < 18 ~ 1,
      between(AGE, 18, 64) ~ 2,
      AGE > 64 ~ 3,
      TRUE ~ 4
    )
  ) |> 
  # Randomization date
  derive_vars_merged(
    dataset_add = ds_ext,
    by_vars = exprs(STUDYID, USUBJID),
    new_vars = exprs(RANDDT = DSSTDT),
    filter_add = DSDECOD == "RANDOMIZED"
  ) |> 
  # End of study date
  derive_vars_merged(
    dataset_add = ds_ext,
    by_vars = exprs(STUDYID, USUBJID),
    new_vars = exprs(EOSDT = DSSTDT),
    filter_add = DSCAT == "DISPOSITION EVENT" & DSDECOD != "SCREEN FAILURE"
  ) |> 
  # End of study status
  mutate(
    EOSSTT = case_when(
      is.na(EOSDT) ~ "ONGOING",
      TRUE ~ "COMPLETED"
    )
  )

# Store ADSL in the data lake
with_transaction(
  create_table(adsl, "adsl"),
  author = "T Gerke",
  commit_message = "Create ADSL dataset",
  commit_extra_info = "Derived from DM, SUPPDM, DS, EX; includes treatment dates, safety flags, age groups"
)

# Variables carried over from DM kept their SDTM labels automatically.
# Derived variables are new, so label them now -- the ADaM way, but stored
# in the catalog rather than a spec sidecar
set_column_comments(
  "adsl",
  TRT01P = "Planned Treatment for Period 01",
  TRT01A = "Actual Treatment for Period 01",
  AGEGR1 = "Pooled Age Group 1",
  AGEGR1N = "Pooled Age Group 1 (N)",
  SAFFL = "Safety Population Flag"
)

# Preview ADSL
get_ducklake_table("adsl") |> 
  select(USUBJID, AGE, AGEGR1, TRT01P, TRTSDT, TRTEDT, SAFFL) |>
  head(10)

## ----build-adae---------------------------------------------------------------
# Read ADSL for merging
adsl <- get_ducklake_table("adsl") |> collect()
ae <- get_ducklake_table("ae") |> collect()

# Build ADAE
adae <- ae |>
  # Merge ADSL variables
  derive_vars_merged(
    dataset_add = adsl,
    new_vars = exprs(TRTSDT, TRTEDT, TRT01A, TRT01P),
    by_vars = exprs(STUDYID, USUBJID)
  ) |>
  # Derive analysis dates
  derive_vars_dt(
    dtc = AESTDTC,
    new_vars_prefix = "AST"
  ) |>
  derive_vars_dt(
    dtc = AEENDTC,
    new_vars_prefix = "AEN",
    date_imputation = "last"
  ) |>
  # Derive treatment-emergent flag
  mutate(
    TRTEMFL = if_else(
      !is.na(ASTDT) & !is.na(TRTSDT) & ASTDT >= TRTSDT,
      "Y",
      NA_character_
    )
  ) |>
  # Derive analysis variables
  mutate(
    AOCCPFL = if_else(AESEQ == min(AESEQ), "Y", NA_character_),
    AOCC01FL = AOCCPFL
  ) |>
  group_by(USUBJID, AEDECOD) |>
  mutate(
    AOCC01FL = if_else(row_number() == 1, "Y", NA_character_)
  ) |>
  ungroup()

# Store ADAE in the data lake
with_transaction(
  create_table(adae, "adae"),
  author = "T Gerke",
  commit_message = "Create ADAE dataset",
  commit_extra_info = "Includes treatment-emergent flags and occurrence flags"
)

# Preview ADAE
get_ducklake_table("adae") |>
  filter(TRTEMFL == "Y") |>
  select(USUBJID, AEDECOD, ASTDT, AESEV, TRTEMFL) |>
  head(10)

## ----build-adpc---------------------------------------------------------------
# Read required datasets
adsl <- get_ducklake_table("adsl") |> collect()
pc <- get_ducklake_table("pc") |> collect()
ex <- get_ducklake_table("ex") |> collect()
vs <- get_ducklake_table("vs") |> collect()

# Get ADSL variables needed
adsl_vars <- exprs(TRTSDT, TRTSDTM, TRT01P, TRT01A)

# Derive PC dates and times
pc_dates <- pc |>
  derive_vars_merged(
    dataset_add = adsl,
    new_vars = adsl_vars,
    by_vars = exprs(STUDYID, USUBJID)
  ) |>
  derive_vars_dtm(
    new_vars_prefix = "A",
    dtc = PCDTC,
    time_imputation = "00:00:00",
    ignore_seconds_flag = FALSE
  ) |>
  derive_vars_dtm_to_dt(exprs(ADTM)) |>
  derive_vars_dtm_to_tm(exprs(ADTM)) |>
  derive_vars_dy(reference_date = TRTSDT, source_vars = exprs(ADT)) |>
  mutate(
    EVID = 0,
    NFRLT = if_else(PCTPTNUM < 0, 0, PCTPTNUM)
  )

# Process exposure records
ex_dates <- ex |>
  derive_vars_merged(
    dataset_add = adsl,
    new_vars = adsl_vars,
    by_vars = exprs(STUDYID, USUBJID)
  ) |>
  filter(EXDOSE > 0) |>
  derive_vars_dtm(
    new_vars_prefix = "AST",
    dtc = EXSTDTC,
    time_imputation = "00:00:00"
  ) |>
  mutate(
    EVID = 1,
    NFRLT = 24 * VISITDY
  ) |>
  derive_vars_dtm_to_dt(exprs(ASTDTM))

# Combine PC and EX
adpc <- bind_rows(pc_dates, ex_dates) |>
  arrange(STUDYID, USUBJID, ADTM) |>
  mutate(
    PARAMCD = coalesce(PCTESTCD, "DOSE"),
    AVAL = case_when(
      EVID == 1 ~ EXDOSE,
      PCSTRESC == "<BLQ" & NFRLT == 0 ~ 0,
      PCSTRESC == "<BLQ" & NFRLT > 0 ~ 0.5 * PCLLOQ,
      TRUE ~ PCSTRESN
    ),
    PARAM = case_when(
      PARAMCD == "XAN" ~ "Xanomeline Concentration",
      PARAMCD == "DOSE" ~ "Xanomeline Dose"
    )
  )

# Store ADPC in the data lake
with_transaction(
  create_table(adpc, "adpc"),
  author = "T Gerke",
  commit_message = "Create ADPC dataset",
  commit_extra_info = "PK concentrations with dosing records for NCA"
)

# Preview ADPC
get_ducklake_table("adpc") |>
  filter(PARAMCD == "XAN") |>
  select(USUBJID, ADT, PCTPT, AVAL, PARAM) |>
  head(10)

## ----store-define-------------------------------------------------------------
# Example: Store define.xml content
# In practice, you might read this from a file generated by your metadata system
define_xml <- '<?xml version="1.0" encoding="UTF-8"?>
<ODM xmlns="http://www.cdisc.org/ns/odm/v1.3">
  <!-- Define.xml content for ADSL, ADAE, ADPC datasets -->
</ODM>'

# Create a table for regulatory documents
regulatory_docs <- tibble(
  doc_type = "define.xml",
  doc_version = "1.0",
  content = define_xml,
  created_date = Sys.Date(),
  description = "Dataset and variable metadata for regulatory submission"
)

with_transaction(
  create_table(regulatory_docs, "regulatory_documents"),
  author = "T Gerke",
  commit_message = "Add define.xml metadata"
)

get_ducklake_table("regulatory_documents")

## ----store-arm----------------------------------------------------------------
# Example: Store Analysis Results Metadata (ARM)
arm_content <- tibble(
  analysis_id = "DEMO01",
  analysis_name = "Demographics Table",
  dataset_used = "ADSL",
  program_name = "t_demographics.R",
  output_file = "t_demographics.rtf",
  analysis_date = Sys.Date()
)

# Add to or update regulatory documents table
with_transaction(
  get_ducklake_table("regulatory_documents") |>
    collect() |>
    mutate(content = as.character(content)) |>
    bind_rows(
      tibble(
        doc_type = "ARM",
        doc_version = "1.0",
        content = as.character(jsonlite::toJSON(arm_content)),
        created_date = Sys.Date(),
        description = "Analysis Results Metadata"
      )
    ) |>
    distinct(doc_type, doc_version, .keep_all = TRUE) |>
    replace_table("regulatory_documents"),
  author = "T Gerke",
  commit_message = "Add ARM metadata"
)

# Example: Store Analysis Results Data (ARD)
ard_content <- tibble(
  analysis_id = "DEMO01",
  row_type = "header",
  row_label = "Age (years)",
  treatment = c("Placebo", "Xanomeline Low", "Xanomeline High"),
  n = c(86, 84, 84),
  mean = c(75.2, 75.7, 74.4),
  sd = c(8.59, 7.89, 7.89)
)

with_transaction(
  get_ducklake_table("regulatory_documents") |>
    collect() |>
    mutate(content = as.character(content)) |>
    bind_rows(
      tibble(
        doc_type = "ARD",
        doc_version = "1.0",
        content = as.character(jsonlite::toJSON(ard_content)),
        created_date = Sys.Date(),
        description = "Analysis Results Data for Demographics Table"
      )
    ) |>
    distinct(doc_type, doc_version, .keep_all = TRUE) |>
    replace_table("regulatory_documents"),
  author = "T Gerke",
  commit_message = "Add demographics ARD"
)

## ----store-specs, eval=FALSE--------------------------------------------------
# # Example: Store ADSL specifications
# adsl_spec <- tibble(
#   dataset = "ADSL",
#   variable = c("USUBJID", "AGE", "AGEGR1", "TRT01P", "SAFFL"),
#   label = c(
#     "Unique Subject Identifier",
#     "Age",
#     "Age Group 1",
#     "Planned Treatment",
#     "Safety Population Flag"
#   ),
#   type = c("text", "num", "text", "text", "text"),
#   length = c(20, 8, 10, 40, 1),
#   derivation = c("DM.USUBJID", "DM.AGE", "Derived from AGE", "DM.ARM", "Derived")
# )
# 
# with_transaction(
#   create_table(adsl_spec, "dataset_specifications"),
#   author = "T Gerke",
#   commit_message = "Add ADSL specifications"
# )
# 
# # Query specifications when needed
# get_ducklake_table("dataset_specifications") |>
#   filter(dataset == "ADSL")

## ----demonstrate-relational---------------------------------------------------
# Traditional approach: Load multiple files, manually join
# adsl <- read_xpt("adsl.xpt")
# ex <- read_xpt("ex.xpt")
# ae <- read_xpt("ae.xpt")
# result <- adsl |> left_join(ex, ...) |> left_join(ae, ...)

# DuckLake approach: Query across related tables directly using dplyr
# Complex cross-domain query without loading all data
adsl_tbl <- get_ducklake_table("adsl")
ex_tbl <- get_ducklake_table("ex")
ae_tbl <- get_ducklake_table("ae")

adsl_tbl |>
  filter(SAFFL == "Y") |>
  left_join(ex_tbl, by = "USUBJID") |>
  left_join(ae_tbl, by = "USUBJID") |>
  group_by(USUBJID, AGE, TRT01P) |>
  summarise(
    n_aes = n_distinct(AESEQ, na.rm = TRUE),
    total_dose = sum(EXDOSE, na.rm = TRUE),
    .groups = "drop"
  ) |>
  arrange(desc(n_aes)) |>
  head(10) |>
  collect()

## ----warehouse-benefits-------------------------------------------------------
# 1. Single source of truth - all datasets in one repository
# List all tables in the data lake
DBI::dbListTables(get_ducklake_connection())

# 2. Efficient filtering before loading into R
# Only load subjects with adverse events
get_ducklake_table("ae") |>
  filter(AESEV == "SEVERE") |>
  distinct(USUBJID)

# 3. Aggregations performed at database level
get_ducklake_table("adae") |>
  filter(TRTEMFL == "Y") |>
  group_by(TRT01A, AESEV) |>
  summarise(
    n_events = n(),
    n_subjects = n_distinct(USUBJID),
    .groups = "drop"
  )

# 4. Joins across SDTM and ADaM layers
# Example: Find date discrepancies between SDTM and ADaM
ae_sdtm <- get_ducklake_table("ae") |>
  select(USUBJID, AESEQ, ae_date = AESTDTC, ae_term = AEDECOD)

adae_adam <- get_ducklake_table("adae") |>
  select(USUBJID, AESEQ, adae_date = ASTDT, adae_term = AEDECOD)

ae_sdtm |>
  inner_join(adae_adam, by = c("USUBJID", "AESEQ")) |>
  # Convert SDTM character date to comparable format for filtering
  mutate(ae_date_comparable = substr(ae_date, 1, 10)) |>
  filter(ae_date_comparable != as.character(adae_date)) |>
  select(
    USUBJID,
    sdtm_start_date = ae_date,
    adam_start_date = adae_date,
    sdtm_term = ae_term,
    adam_term = adae_term
  )
# Note: This returns 0 rows with clean pharmaversesdtm data,
# but demonstrates how to check for data quality issues across layers

## ----explore-structure--------------------------------------------------------
# List all tables in the data lake
DBI::dbListTables(get_ducklake_connection())

# View snapshot history for key tables
metadata_tables <- c("dm", "ex", "ae", "pc", 
                     "adsl", "adae", "adpc")

# Collect snapshots for all tables
purrr::map_dfr(metadata_tables, ~{
  list_table_snapshots(.x) |>
    mutate(table = .x, .before = 1)
}) |>
  select(table, snapshot_id, snapshot_time, changes)

# Check all ADAE subjects exist in ADSL
adae_tbl <- get_ducklake_table("adae")
adsl_tbl <- get_ducklake_table("adsl")

integrity_check <- adae_tbl |>
  anti_join(adsl_tbl, by = "USUBJID") |>
  summarise(orphaned_records = n()) |>
  collect()

# ADAE records without ADSL subject:
integrity_check |> pull(orphaned_records)

## ----demonstrate-versioning---------------------------------------------------
# Add new derived columns using dplyr syntax
# replace_table() handles the DROP/CREATE cycle internally
with_transaction(
  get_ducklake_table("adsl") |>
    mutate(
      AGE65FL = if_else(AGE >= 65, "Y", "N"),
      AGECAT = case_when(
        AGE < 65 ~ "<65",
        AGE >= 65 & AGE < 75 ~ "65-74",
        AGE >= 75 ~ ">=75",
        TRUE ~ NA_character_
      )
    ) |>
    replace_table("adsl"),
  author = "T Gerke",
  commit_message = "Add age categorization vars"
)

# View version history - should now show 2 snapshots
list_table_snapshots("adsl")

# Verify new columns exist
get_ducklake_table("adsl") |>
  select(USUBJID, AGE, AGE65FL, AGECAT) |>
  head(5)

## ----iterative-workflow-------------------------------------------------------
# Iteration 1: First attempt (creates snapshot v2)
with_transaction(
  get_ducklake_table("adsl") |>
    mutate(AGECAT_TEST = case_when(
      AGE < 50 ~ "Young",
      AGE >= 50 ~ "Older"
    )) |>
    replace_table("adsl"),
  author = "T Gerke",
  commit_message = "Test age categories v1"
)

# Iteration 2: Refinement (creates snapshot v3)
with_transaction(
  get_ducklake_table("adsl") |>
    mutate(AGECAT_TEST = case_when(
      AGE < 40 ~ "18-39",
      AGE < 65 ~ "40-64",
      AGE >= 65 ~ "65+"
    )) |>
    replace_table("adsl"),
  author = "T Gerke",
  commit_message = "Refine age categories v2"
)

# Iteration 3: Final version (creates snapshot v4)
with_transaction(
  get_ducklake_table("adsl") |>
    mutate(
      AGECAT_TEST = NULL,
      AGECAT2 = case_when(
        AGE < 40 ~ "18-39",
        AGE < 65 ~ "40-64",
        AGE >= 65 ~ "65+",
        TRUE ~ "Missing"
      )
    ) |>
    replace_table("adsl"),
  author = "T Gerke",
  commit_message = "Finalize age categories"
)

# Complete audit trail available
snapshots <- list_table_snapshots("adsl")
snapshots  # Shows all iterations with snapshot metadata

# Each row represents a point in time you can restore to
# - snapshot_id: Unique identifier for this version
# - snapshot_time: When this version was created
# - changes: What operations created this snapshot

# Time-travel to specific snapshots using snapshot_id
# Use actual snapshot IDs from the list (first, second, and last)
snapshot_ids <- snapshots$snapshot_id
adsl_v1 <- get_ducklake_table_version("adsl", snapshot_ids[1])
adsl_v2 <- get_ducklake_table_version("adsl", snapshot_ids[2])
adsl_final <- get_ducklake_table_version("adsl", snapshot_ids[length(snapshot_ids)])

# Or use snapshot times for time-travel
# Note: Add 1 second to ensure we query AFTER the snapshot was created
adsl_asof <- get_ducklake_table_asof("adsl", snapshots$snapshot_time[2] + 1)

# Compare columns across iterations
colnames(adsl_v1 |> collect())  # Initial version
colnames(adsl_final |> collect())  # Final version with all derivations

## ----demonstrate-time-travel--------------------------------------------------
# Get the current version
adsl_current <- get_ducklake_table("adsl")

# Get the version history for adsl
versions <- list_table_snapshots("adsl")
print(versions)

# Get data from the first snapshot version
first_snapshot_id <- versions |>
  slice(1) |>
  pull(snapshot_id)

adsl_v1 <- get_ducklake_table_version(
  table_name = "adsl",
  version = first_snapshot_id
)

# Compare versions - earlier version shouldn't have derived variables added later
adsl_v1 |> collect()
adsl_current |> collect()

## ----demonstrate-transactions-------------------------------------------------
# Add ANALYSISFL to both ADSL and ADAE in a single atomic operation
# with_transaction() automatically handles rollback on error
with_transaction({
  # First, add the flag to ADSL
  get_ducklake_table("adsl") |>
    mutate(ANALYSISFL = if_else(SAFFL == "Y" & !is.na(TRTSDT), "Y", "N")) |>
    replace_table("adsl")  # Creates versioned snapshot
  
  # Then propagate to ADAE by joining
  adsl_flags <- get_ducklake_table("adsl") |>
    select(USUBJID, ANALYSISFL)
  
  get_ducklake_table("adae") |>
    select(-any_of("ANALYSISFL")) |>  # Remove if exists
    left_join(adsl_flags, by = "USUBJID") |>
    replace_table("adae")  # Creates versioned snapshot
  
  # Both updates succeed together
  cat("Both tables updated successfully\n")
}, author = "T Gerke", commit_message = "Add analysis flag")

## ----update-records-----------------------------------------------------------
# Update a specific record with versioning
with_transaction(
  get_ducklake_table("adae") |>
    mutate(
      AESEV = if_else(
        USUBJID == "01-701-1015" & AESEQ == 1,
        "SEVERE",
        AESEV
      )
    ) |>
    replace_table("adae"),  # Creates versioned snapshot
  author = "T Gerke",
  commit_message = "Correct AE severity"
)

# Verify the update
get_ducklake_table("adae") |>
  filter(USUBJID == "01-701-1015", AESEQ == 1) |>
  select(USUBJID, AEDECOD, AESEV)

## ----analysis-examples--------------------------------------------------------
# Example 1: Subject disposition summary
get_ducklake_table("adsl") |>
  count(EOSSTT, TRT01P) |>
  arrange(TRT01P, EOSSTT)

# Example 2: Treatment-emergent AE summary by severity
get_ducklake_table("adae") |>
  filter(TRTEMFL == "Y") |>
  count(TRT01A, AESEV) |>
  arrange(TRT01A, AESEV)

# Example 3: PK concentration profile
get_ducklake_table("adpc") |>
  filter(PARAMCD == "XAN", EVID == 0) |>
  group_by(NFRLT) |>
  summarise(
    n = n(),
    mean_conc = mean(AVAL, na.rm = TRUE),
    sd_conc = sd(AVAL, na.rm = TRUE)
  ) |>
  arrange(NFRLT)

# Example 4: Cross-domain analysis: AEs by age group
get_ducklake_table("adae") |>
  filter(TRTEMFL == "Y") |>
  left_join(
    get_ducklake_table("adsl") |>
      select(USUBJID, AGEGR1, TRT01A),
    by = "USUBJID"
  ) |>
  count(AGEGR1, TRT01A.x) |>
  arrange(AGEGR1, TRT01A.x)

## ----audit-trail--------------------------------------------------------------
# Generate audit report for ADSL
audit_report <- list_table_snapshots("adsl")
audit_report

# Get table metadata from DuckLake system tables
adsl_table_meta <- get_metadata_table("ducklake_table") |>
  filter(table_name == "adsl") |>
  collect()
adsl_table_meta

# Export audit information
audit_export <- audit_report |>
  mutate(
    table_name = "adsl",
    dataset_label = "Subject-Level Analysis Dataset"
  )
audit_export

## ----pre-archival-checkpoint, eval = FALSE------------------------------------
# # Materialise any inlined data to Parquet before archival
# checkpoint_ducklake()

## ----cleanup------------------------------------------------------------------
detach_ducklake()