The skeleton pipeline

The skeleton pipeline

This vignette describes how swereg builds per-batch skeleton files in production, and how it rebuilds them incrementally. To create a small skeleton and add some diagnoses by hand, read vignette("skeleton-create") first. This vignette is for full-registry pipelines, where the total build time is several days.

The problem

A production swereg pipeline on Swedish registry data typically has two main costs:

1. Rawbatch creation: split the raw registry files into per-person-batch subsets, and save each as .qs2. At national-registry scale this typically takes many hours of I/O.
2. Skeleton processing: for each batch, build a person-week time grid, apply censoring, add demographics from LISA, classify exposure from LMED, and match ICD-10 / ATC / operation codes across the subset. This step typically takes most of the pipeline time.

Together these give a multi-day first build for national-scale studies. A bare data.table holds no provenance to compare against, so without one an edit to one ICD-10 code forces a full rebuild.

The phased pipeline makes the rebuild cost proportional to the size of the change. An edit re-runs only the phases it invalidates.

Four phases, and the order they run in

Per batch, RegistryStudy$process_skeletons() runs four phases:

framework  ->  trim  ->  codes  ->  randvars

Each phase has its own invalidation rule, so an edit re-runs only what it affects. The sections below describe the phases in that order.

Phase 1: framework

A single user-supplied function, registered via $register_framework(fn), with signature function(batch_data, config). It returns a fresh data.table containing the base time grid plus structural censoring (immigration, emigration, death). Every later phase needs it.

Invalidation: full rebuild whenever digest::digest(list(body(fn), formals(fn)), "xxhash64") changes. Framework edits are rare, because the time grid seldom changes, so a full rebuild on every hash change is acceptable. When phase 1 re-runs, swereg discards every later phase and re-applies it from scratch on the fresh base.

Phase 1b: trim

At most one user-supplied function, registered via $register_trim(fn), with signature function(skeleton, batch_data, config). It returns a data.table, and $process_skeletons() rebinds the skeleton to what it returns. A study that registers no trim skips this phase.

This is the one place in the pipeline that MAY delete skeleton rows. A code entry or a randvars step that changes the row count stops the run, and the error names the registration to edit. Phase 1b runs before the code registry, so every later phase sees the rows the trim leaves behind.

study$register_trim(function(skeleton, batch_data, config) {
  skeleton[isoyear >= 2010]
})

Invalidation: full rebuild whenever the trim’s identity changes. Three changes rebuild the base of every batch: a new trim, a removed trim, and an edit to the trim’s body. swereg cannot undo a row deletion, so it must start again from a fresh base.

The trim runs on a rebuild and at no other time. It therefore always reads a fresh base, so it MAY delete a fixed count or a fraction rather than a predicate. An edit to a code entry or to a randvars step rebuilds nothing and re-runs no trim.

A Skeleton records the trim in trim_fn_hash, not in phase_order. phase_order names the three phases whose order a swereg release can change. The trim always runs immediately after the framework.

$register_trim() stops on a second call. A study has one framework by construction, so $register_framework() overwrites instead. A second trim registration means that the script deletes rows in two places.

Phase 2: code registry

A list of code registrations, each added via $register_codes() (primary) or $register_derived_codes() (derived). Each entry adds one or more columns to the skeleton. swereg predicts the shape of those columns from metadata. It does not run fn to find it.

Invalidation: per-entry fingerprint diff. Skeleton$sync_with_registry() computes to_drop = stored_fingerprints - current_fingerprints and to_add = current_fingerprints - stored_fingerprints, drops stale entries’ columns, then re-applies new entries in registration order. When you change one code in one entry, every other entry’s fingerprint stays the same, so only that entry re-runs in phase 2. A code entry edit also replays every phase-3 step.

Phase 3: randvars

An ordered named list of user-supplied functions, each registered via $register_randvars(name, fn). Each has signature function(skeleton, batch_data, config). A step adds columns to the skeleton. It MUST NOT modify or delete an existing column: the rewind tracks only the columns a step records.

A randvars step MAY read a code registry column, because phase 2 runs before phase 3. This is the reason for the phase order. A step MUST NOT change the row count. Rewind-and-replay drops columns, and it cannot restore rows. Register a row filter as the phase-1b trim instead.

Registration order is execution order. An edit to one step replays that step and everything downstream of it. It leaves upstream steps untouched.

Invalidation: divergence-point rewind-and-replay. swereg compares the stored (name, hash) sequence with the current one. It finds the first position where the name or the hash differs. It drops the stored added_columns of every step from that position on, then replays the current steps from that position. An add, a removal, an edit and a reorder each change the name sequence or the hash sequence, so this one rule handles all four.

Each step’s hash folds in the framework hash, the trim identity and every code registry fingerprint. A code edit therefore changes every step’s hash. The divergence point is step 1, and the whole sequence replays against the re-applied code columns.

A typical phase 3 registration block:

study$register_randvars("demographics",        rv_demographics)
study$register_randvars("exposure",            rv_exposure)
study$register_randvars("baseline_exclusion",  rv_baseline_exclusion)

Exclusion often depends on exposure classification, so baseline_exclusion often reads a column that exposure produces. An edit to exposure then puts the divergence point at step 2, and steps 2 and 3 both replay. Only the registration order records this dependency. swereg has no explicit dependency graph.

swereg counts each randvars column in the batch meta, and the count stops at 100 values. The meta file next to each skeleton counts a logical randvars column under its own name. It counts a character or factor column once per value the column holds, not once per level a factor declares. A column that holds more than 100 distinct values is skipped, and its name goes into the meta’s randvars_counts_skipped. $compute_summary() reads these counts, so a skipped column contributes no summary row.

An exposure column or a demographic column stays far below the cap. An identifier column or a free-text column does not. Use $compute_population() for a value distribution of that size.

The Skeleton R6 class

Each per-batch file on disk is a serialized Skeleton object, not a bare data.table. A Skeleton carries its own provenance:

sk <- study$load_skeleton(batch_number = 1L)
sk
#> <Skeleton batch 1>
#>   rows:             8,234,112
#>   cols:             287
#>   framework_hash:   abcd1234efgh
#>   trim_hash:        9f3c77aa1122
#>   phase_order:      framework -> codes -> randvars
#>   randvars steps:   3
#>   applied codes:    127
#>   pipeline_hash:    4d8f99af7b2c

sk$data                          # the underlying data.table
sk$framework_fn_hash             # phase-1 hash
sk$trim_fn_hash                  # phase-1b trim identity
sk$phase_order                   # the order the phases ran in
names(sk$randvars_state)         # applied phase-3 step names in order
length(sk$applied_registry)      # phase-2 entries currently materialized
sk$pipeline_hash()               # rolled-up provenance scalar

Why R6 and not a bare data.table?

The provenance is the state that matters. Without it, you cannot tell if a skeleton is still valid for the current pipeline. You must rebuild from scratch or trust a separate sidecar file. With it, each batch file answers the question locally via sk$pipeline_hash() == study$pipeline_hash().

qs2 files and data.table over-allocation

data.table keeps over-allocated pointer slots (“truelength”) so that an in-place := does not have to reallocate. qs2 does not keep those slots. A table read back from disk therefore has a truelength of 0, and the next := inside a function reallocates it. The function then writes to a shallow copy of its own, and the caller never sees the new column. data.table reports nothing.

swereg::qs2_read() repairs this, and every swereg load path gets the repair. RegistryStudy$load_skeleton() therefore hands you a Skeleton whose $data is already over-allocated:

sk <- study$load_skeleton(1L)
data.table::truelength(sk$data) > ncol(sk$data)   # TRUE

The repair also reaches a data.table nested inside a list or inside an R6 field. ?qs2_read states exactly what it reaches and what it leaves alone.

Read your own loader’s files with swereg::qs2_read(), not qs2::qs_read(), to get the same repair.

The code registry: primary and derived entries

Primary entries via register_codes()

A primary entry ties a code list to a fn (e.g. add_diagnoses, add_cods, add_rx) and a set of raw-data groups to apply it to. The optional combine_as runs fn once more on the rbind of all groups to make a combined column.

study$register_codes(
  codes      = list(e11 = c("E11"), vte = c("I26", "I80")),
  fn         = swereg::add_diagnoses,
  groups     = list(ov = "outpatient", sv = "inpatient"),
  combine_as = "os"
)

swereg writes one column for each (group_prefix, code_name) pair: ov_e11, sv_e11, ov_vte, sv_vte. Because combine_as = "os" is set, it writes two more columns: os_e11 = ov_e11 | sv_e11 and os_vte = ov_vte | sv_vte. swereg computes them by re-running add_diagnoses on the rbind of outpatient and inpatient rawbatch data. It does not OR the columns.

Combine hospital and DORS codes with register_derived_codes()

One ICD-10 column that covers hospital data and DORS needs two different functions. add_diagnoses reads hospital data. It searches the hdia/dia*/ekod*/icd* columns, and the cause-of-death registry has none of them. DORS uses ulorsak (underlying cause) and morsak* (contributing causes), which add_cods reads.

combine_as cannot combine the two, because it reruns the same fn on the rbind data. register_derived_codes() ORs together columns that different registrations produce:

# 1. Hospital (OV + SV) -> ov_*, sv_*, os_*
study$register_codes(
  codes      = ICD10_CODES,
  fn         = swereg::add_diagnoses,
  groups     = list(ov = "outpatient", sv = "inpatient"),
  combine_as = "os"
)

# 2. DORS underlying cause -> dorsu_*
study$register_codes(
  codes   = ICD10_CODES,
  fn      = swereg::add_cods,
  fn_args = list(cod_type = "underlying"),
  groups  = list(dorsu = "dors")
)

# 3. DORS contributing causes -> dorsm_*
study$register_codes(
  codes   = ICD10_CODES,
  fn      = swereg::add_cods,
  fn_args = list(cod_type = "multiple"),
  groups  = list(dorsm = "dors")
)

# 4. Derived: osd_<nm> = os_<nm> | dorsu_<nm> | dorsm_<nm>
study$register_derived_codes(
  codes = ICD10_CODES,
  from  = c("os", "dorsu", "dorsm"),
  as    = "osd"
)

A derived entry does not read rawbatch data. It iterates its own codes list, looks up get("<from[1]>_<nm>"), get("<from[2]>_<nm>"), etc. on the skeleton, ORs them, and writes <as>_<nm>. The phase 2 sync runs the entries in registration order and does not reorder them. Register every primary entry that a derived entry reads before the derived entry. $register_derived_codes() does not check the order. A missing source column raises an error when the sync applies the derived entry.

Fingerprint cascade for derived entries

Derived entries fold the fingerprints of their upstream primary entries into their own fingerprint. An edit to any upstream field (fn_args, groups, or one code pattern) therefore replays the derived entry, and you do not touch the derived entry. If you change cod_type = "underlying" to cod_type = "both" on the DORS primary, the next process_skeletons() run drops the derived entry’s columns. It then rewrites them from the new dorsu output on every batch, in order, and rebuilds nothing else.

code_registry_fingerprints() implements the cascade as a two-pass walk:

1. Compute every primary entry’s fingerprint. It depends on codes, label, groups, fn_args, combine_as and the hash of the entry’s fn.
2. For each derived entry, walk entries registered before it, collect the primary fingerprints whose groups prefixes or combine_as appear in the derived’s from, and hash (kind="derived", codes, from, as, upstream_fps).

Registering the pipeline

A complete runner script has the shape:

study <- swereg::RegistryStudy$new(
  data_rawbatch_dir = c("/mnt/shared/rawbatch/", "C:/shared/rawbatch/"),
  data_skeleton_dir = c("/mnt/shared/skeleton/", "C:/shared/skeleton/"),
  data_raw_dir      = c("/mnt/shared/raw/",      "C:/shared/raw/"),
  group_names       = c("lmed", "inpatient", "outpatient",
                        "cancer", "dors", "other")
)

# Phase 1
study$register_framework(my_framework_fn)

# Phase 1b (optional; the only place rows may be deleted)
study$register_trim(my_trim_fn)

# Phase 2: primaries first, then derived
study$register_codes(ICD10_CODES, swereg::add_diagnoses,
  groups = list(ov = "outpatient", sv = "inpatient"), combine_as = "os")
study$register_codes(ICD10_CODES, swereg::add_cods,
  fn_args = list(cod_type = "underlying"),
  groups = list(dorsu = "dors"))
study$register_codes(ICD10_CODES, swereg::add_cods,
  fn_args = list(cod_type = "multiple"),
  groups = list(dorsm = "dors"))
study$register_derived_codes(ICD10_CODES,
  from = c("os", "dorsu", "dorsm"), as = "osd")

# Phase 3 (registration order = execution order)
study$register_randvars("demographics",        rv_demographics)
study$register_randvars("exposure",            rv_exposure)
study$register_randvars("baseline_exclusion",  rv_baseline_exclusion)

# ... rawbatch creation (one-time) ...

study$process_skeletons(n_workers = 4L)

Register the framework, the trim, the code registry and the randvars steps on study BEFORE you call $process_skeletons(). $process_skeletons() has no callback argument. The study itself holds the pipeline, and no call passes it in.

Batches that keep matched sets together

$set_ids() shuffles the IDs before it splits them into batches. So the case and the controls of one matched set usually go into different batches. Two methods set the batch composition directly.

$set_batches() takes one vector of IDs per batch, in the given order. An ID MAY be in more than one batch. $save_rawbatch() then writes every row of that ID into each batch that holds it. The population and summary counts sum over the batches, so they count that person once per batch.

$set_matched_sets() takes a roster with one row per membership:

roster <- data.table::data.table(
  lopnr       = c(101, 202, 303, 404, 202),
  set_id      = c("a", "a", "a", "b", "b"),
  role        = c("case", "control", "control", "case", "control"),
  skeleton_id = c(101, 202000101, 303000101, 404, 202000404)
)
study$set_matched_sets(roster, sets_per_batch = 50L)

The method keeps each set in one batch. It packs the sets in order of first appearance of set_id, sets_per_batch sets per batch.

Person 202 is a control in sets a and b. $save_rawbatch() writes each rawbatch row of person 202 twice, once with lopnr 202000101 and once with 202000404. The other columns of the rows do not change. A framework that builds the skeleton from these rows gives each membership its own id. Phase 2 then gives both memberships the codes of person 202.

The roster has three rules:

• One skeleton_id MUST map to one person.
• A skeleton_id MUST NOT repeat within one batch. It MAY repeat across batches.
• The role column MUST be character.

$save_rawbatch() also writes the roster rows of each batch as the rawbatch group "roster", with lopnr and skeleton_id unchanged. A phase function reads them as batch_data[["roster"]]. This randvars step adds the set and the role to the skeleton:

rv_roster <- function(skeleton, batch_data, config) {
  roster <- batch_data[["roster"]]
  skeleton[roster, on = c(id = "skeleton_id"),
    `:=`(set_id = i.set_id, role = i.role)]
  return(skeleton)
}

Do not change the roster while rawbatch or skeleton files are on disk. The recorded composition includes the roster, and $set_ids(), $set_batches() and $set_matched_sets() stop when the new composition differs from it. Delete the files with $delete_rawbatches() and $delete_skeletons() first.

The reload-clobber anti-pattern

This pattern fails when you run the pipeline more than once:

# WRONG
study <- swereg::RegistryStudy$new(...)
study$register_framework(current_framework_fn)     # fresh in-memory
study$register_codes(current_code_list, ...)       # fresh in-memory

if (file.exists(study$meta_file)) {
  study <- swereg::qs2_read(study$meta_file)       # CLOBBER
}

The reload replaces the newly registered pipeline with the stale version on disk. It reverts any edits you just made, with no warning. Instead, copy only the persisted runtime state (IDs, batches, groups_saved) into the new study, and leave its config unchanged:

# CORRECT
if (file.exists(study$meta_file)) {
  study$adopt_runtime_state_from(swereg::qs2_read(study$meta_file))
}

Provenance verification

Each Skeleton can compute its own pipeline_hash() from its stored phase provenance. The study exposes the expected value:

sk <- study$load_skeleton(1L)
sk$pipeline_hash() == study$pipeline_hash()
# FALSE means this batch is definitely stale

Equality is necessary for a synced batch, and it is not sufficient. FALSE means the batch is definitely stale. TRUE means only that nothing changed among the inputs both hashes cover. Those inputs are the framework function, the trim identity, the phase order, the randvars sequence and the code registry fingerprints.

Two inputs sit outside both hashes. The rawbatch data is one: nothing hashes raw content. Whatever a registered function calls or reads is the other. Each hash covers that function’s own body and formals, and follows no call into a helper. Change either one and the two hashes stay equal over a stale skeleton.

Two summary helpers:

# Per-batch table with framework hash, randvars count, code count, saved_at
study$skeleton_pipeline_hashes()

# Errors loudly if any persisted skeleton is out of sync
study$assert_skeletons_consistent()

tteplan_from_spec_and_registrystudy() calls $assert_skeletons_consistent() near its start. The call stops when two batches have different pipeline identities. It also stops when the batches agree with each other but not with the current pipeline of the study. A plan therefore never reads a skeleton that an older pipeline built.

Expected re-run costs

After the first full build:

Edit	Cost
Nothing	Per-batch no-op: load Skeleton, see nothing changed, save. Seconds per batch.
One code pattern in one primary entry	Drop that entry’s columns on all batches and re-apply it: roughly (phase-2 total) / (N primaries). Then replay the whole randvars sequence.
One primary’s fn_args (e.g. cod_type)	Same as above, plus the cascade re-applies any derived entries that depend on it.
One primary’s registered fn body	Same as above. The fingerprint folds in that function’s body and formals.
One phase-3 step’s body	Drop that step’s columns on all batches, replay it and everything downstream of it in the randvars sequence.
The trim function’s body	Full rebuild from scratch, all four phases.
The framework function’s body	Full rebuild from scratch, all four phases. Rare.

An “edit one ICD-10 code” run re-applies that one entry and replays the whole randvars sequence. It never re-runs the framework, the trim, or the other code entries. On a national-registry pipeline with a multi-day first build, this costs much less than a full rebuild.

Summary

• Per-batch skeleton files are Skeleton R6 objects. Each carries the framework hash, the trim identity, the phase order, the ordered randvars state, and the applied code registry fingerprint map.
• Each of the four phases has its own invalidation rule: framework (full rebuild on hash change), trim (full rebuild on identity change), codes (per-entry fingerprint diff), randvars (divergence-point rewind and replay).
• Only the phase-1b trim may delete rows. A code entry or a randvars step that changes the row count stops the run.
• Primary and derived code entries let you combine outputs from different registration functions into one logical column. Derived entries cascade upstream fingerprint changes automatically.
• pipeline_hash() lets any consumer check whether a skeleton matches the current pipeline.
• adopt_runtime_state_from() is the correct reload pattern. Do not reassign the study from disk, because that overwrites the newly registered pipeline.

See also:

• ?RegistryStudy: full method reference
• ?Skeleton: per-batch provenance object
• vignette("skeleton-concept"): the reason for the person-week time grid
• vignette("rowdep-rowind-concept"): variable-type conventions