Skip to contents

Staged, sequential ingestion of wide measurement tables

Source: R/wide_measurement.R
load_wide_measurement.Rd

Unpivoting a wide table (PCORnet vital, flowsheets, device streams) with one giant uncommitted UNION ALL materializes the whole result in DuckDB's append buffer and can exhaust RAM. This function (a) stages the source once into a keyed intermediate table (integer primary key, values already cast to DOUBLE, dates already normalized with vectorized_strptime()) and (b) inserts into measurement one concept at a time, each pass in its own committed transaction, applying per-measurement plausibility bounds. Peak memory is one column's worth of rows and a failed pass loses only that pass. Mirror of omop_etl.wide_measurement.load_wide_measurement in Python.

Usage

load_wide_measurement(
  con,
  source,
  concept_mapping,
  person_col = "PATID",
  date_col = "MEASURE_DATE",
  time_col = "MEASURE_TIME",
  visit_col = "ENCOUNTERID",
  staged_table_name = "_staged_wide_measurement",
  batch_execution_mode = c("sequential", "single"),
  value_plausibility_filters = NULL,
  person_id_is_cdm_id = FALSE,
  measurement_type_concept_id = 32817L,
  target_table = "measurement",
  datetime_formats = NULL,
  datetime_on_error = "null",
  keep_staged = FALSE,
  missing_columns = c("raise", "skip"),
  coalesce_columns = "auto"
)

Arguments

con

DuckDB DBI connection holding the OMOP CDM (person, visit_occurrence, measurement).

source

Table / view name, CSV / TSV / Parquet path, or a parenthesized query.

concept_mapping

Named list source_column -> list(concept_id, loinc, unit_concept_id, unit). A bare number is shorthand for list(concept_id = n). When concept_id is absent the loinc code is resolved through concept / concept_relationship. See DEFAULT_VITAL_CONCEPT_MAPPING.

person_col, date_col, time_col, visit_col

Source column names (case-insensitive); time_col and visit_col may be NULL.

staged_table_name

Name of the intermediate table (integer primary key stage_id).

batch_execution_mode

"sequential" (default; one committed single-concept insert per pass) or "single" (all passes in one transaction, committed at the end; only for small inputs).

value_plausibility_filters

Named list source_column -> c(min, max) of inclusive clinical bounds; either bound may be NA. Out-of-range values are dropped and counted. A min / max inside a mapping entry is used when the column has no filter here. See DEFAULT_VITAL_PLAUSIBILITY_FILTERS.

person_id_is_cdm_id

Treat person_col as person.person_id instead of the source patient id (matched to person.person_source_value).

measurement_type_concept_id

Type concept stamped on every row.

target_table

Destination (a measurement-shaped table).

datetime_formats, datetime_on_error

Passed to vectorized_strptime().

keep_staged

Keep the staging table after loading (default: drop it).

missing_columns

"raise" (default) if a mapped column is absent from the source, or "skip" to ignore it (useful with DEFAULT_VITAL_CONCEPT_MAPPING on partial extracts).

coalesce_columns

Source columns that carry the same measure (e.g. PCORnet BMI and ORIGINAL_BMI) and must yield ONE measurement row per source row instead of one each. list(c("BMI", "ORIGINAL_BMI")) lists a precedence group: the first column with a value inside its plausibility bounds wins, later columns only fill gaps. "auto" (default) builds a group, in concept_mapping order, from every set of columns sharing a concept_id; NULL disables coalescing (one row per column, as in 0.5.7). Every column of a group must map to the same concept_id.

Value

A data frame with one row per source column: source_column, concept_id, rows_non_null, rows_out_of_range, rows_unmatched_person, rows_inserted, coalesced_into. A column folded into an earlier one reports rows_inserted = 0 and the primary column's name in coalesced_into; the primary's counts cover the whole group.