config

MESSY (MEDS-Extract Specification Syntax YAML) config parsing.

The MESSY config is a small DSL for extracting MEDS events from raw source tables. It resolves to a list of :class:TableConfig entries — each owning a list of :class:EventConfig entries plus the subject_id expression, derived columns, and join target that apply to the whole table.

All parsing, validation, and polars expression construction happen through :class:MessyConfig. Stages should call :meth:MessyConfig.load once and then consume the parsed class via its methods (table.extract_events, config.needed_source_columns, etc.) — they should never traverse the raw dict themselves.

CompiledMetadataBlock dataclass

One event’s _metadata entry for one metadata prefix, compiled and classified.

exprs maps every produced column to its compiled dftly node, in declared order. key_cols are the produced columns whose names match the declaring event’s code-referenced components — the join keys, sorted. output_cols are all other produced columns, in declared order — the metadata attached to matched codes. code_template is the declaring event’s code expression string — the provenance value extract_code_metadata stamps on every extracted metadata row.

Source code in MEDS_extract/config.py

@dataclass(frozen=True)
class CompiledMetadataBlock:
    """One event's ``_metadata`` entry for one metadata prefix, compiled and classified.

    ``exprs`` maps every produced column to its compiled dftly node, in declared
    order. ``key_cols`` are the produced columns whose names match the declaring
    event's code-referenced components — the join keys, sorted. ``output_cols`` are
    all other produced columns, in declared order — the metadata attached to matched
    codes. ``code_template`` is the declaring event's code expression string — the
    provenance value ``extract_code_metadata`` stamps on every extracted metadata row.
    """

    exprs: dict[str, NodeBase]
    key_cols: tuple[str, ...]
    output_cols: tuple[str, ...]
    code_template: str

    @cached_property
    def referenced_columns(self) -> frozenset[str]:
        """Raw metadata-table columns referenced by the compiled expressions."""
        cols: set[str] = set()
        for node in self.exprs.values():
            cols.update(node.referenced_columns)
        return frozenset(cols)

referenced_columns cached property

Raw metadata-table columns referenced by the compiled expressions.

EtlConfig dataclass

The parsed reserved etl: block of a MESSY file — one section of :class:MessyConfig.

Constructed by :meth:MessyConfig.parse in full-document context (never loaded from a file independently). The block is reserved exactly like sources:: stripped before event-table parsing and consumed only by the meds-extract-run CLI. Unlike sources: it carries no credentials, so it is neither redacted from log output nor from :meth:MessyConfig.save copies. Every key is optional — a registered dataset whose sources: block declares dataset_version needs no etl: block at all:

• dataset_name: defaults to the registered MEDS_extract.pipelines name (see :meth:MessyConfig.dataset_name); required here only for pkg:///path-resolved specs.
• raw_dataset_version: fallback for specs whose sources: block declares no dataset_version (see :meth:MessyConfig.raw_version_for).
• Curated stage options — real stage-parameter names, no aliases, each mapped internally onto its owning stage in the canonical :attr:DEFAULT_PIPELINE. The stage sequence itself is not configurable: nonstandard pipeline shapes are served by writing a custom pipeline YAML and running MEDS_transform-pipeline directly.

Examples:

An empty block is valid; the canonical pipeline is implied, and curated options land on their owning stage:

>>> EtlConfig.parse({}).stages_container() == list(EtlConfig.DEFAULT_PIPELINE)
True
>>> etl = EtlConfig.parse({"n_subjects_per_shard": 1000, "do_dedup_text_and_numeric": False})
>>> etl.stages_container()[:4]
['convert_to_parquet',
 {'split_and_shard_subjects': {'n_subjects_per_shard': 1000}},
 'convert_to_subject_sharded',
 {'convert_to_MEDS_events': {'do_dedup_text_and_numeric': False}}]

Unknown keys are rejected by name, listing the allowed set; options are type-validated against their stage’s contract (bools are rejected where ints are expected — YAML true must not pass as 1); version/name values must be strings (an unquoted YAML 3.1 parses as a float and gets a targeted quote-it message):

>>> EtlConfig.parse({"raw_dataset_version": "1", "pipeline": ["convert_to_parquet"]})
Traceback (most recent call last):
    ...
ValueError: etl: block contains unknown key(s) ['pipeline']. Allowed keys:
['dataset_name', 'description_separator', 'do_dedup_text_and_numeric',
'external_splits_json_fp', 'n_subjects_per_shard', 'raw_dataset_version',
'split_fracs']. The stage sequence itself is not configurable here —
nonstandard pipeline shapes are served by writing a custom pipeline YAML and running
`MEDS_transform-pipeline` directly (see the README's "Custom pipeline shapes").
>>> EtlConfig.parse({"n_subjects_per_shard": True})
Traceback (most recent call last):
    ...
ValueError: etl.n_subjects_per_shard (a `split_and_shard_subjects` option) must be a
positive int, got bool (True).
>>> EtlConfig(raw_dataset_version=3.1)
Traceback (most recent call last):
    ...
ValueError: etl.raw_dataset_version must be a non-empty string, got float (3.1).
Quote the version in YAML: raw_dataset_version: "3.1".

Source code in MEDS_extract/config.py

@dataclass(frozen=True)
class EtlConfig:
    """The parsed reserved ``etl:`` block of a MESSY file — one section of :class:`MessyConfig`.

    Constructed by :meth:`MessyConfig.parse` in full-document context (never loaded
    from a file independently). The block is reserved exactly like ``sources:``:
    stripped before event-table parsing and consumed only by the ``meds-extract-run``
    CLI. Unlike ``sources:`` it carries no credentials, so it is neither redacted
    from log output nor from :meth:`MessyConfig.save` copies. Every key is optional —
    a registered dataset whose ``sources:`` block declares ``dataset_version`` needs
    no ``etl:`` block at all:

    - ``dataset_name``: defaults to the registered ``MEDS_extract.pipelines`` name
      (see :meth:`MessyConfig.dataset_name`); required here only for
      ``pkg://``/path-resolved specs.
    - ``raw_dataset_version``: **fallback** for specs whose ``sources:`` block
      declares no ``dataset_version`` (see :meth:`MessyConfig.raw_version_for`).
    - Curated stage options — real stage-parameter names, no aliases, each mapped
      internally onto its owning stage in the canonical :attr:`DEFAULT_PIPELINE`.
      The stage sequence itself is not configurable: nonstandard pipeline shapes
      are served by writing a custom pipeline YAML and running
      ``MEDS_transform-pipeline`` directly.

    Examples:
        An empty block is valid; the canonical pipeline is implied, and curated
        options land on their owning stage:

        >>> EtlConfig.parse({}).stages_container() == list(EtlConfig.DEFAULT_PIPELINE)
        True
        >>> etl = EtlConfig.parse({"n_subjects_per_shard": 1000, "do_dedup_text_and_numeric": False})
        >>> etl.stages_container()[:4]
        ['convert_to_parquet',
         {'split_and_shard_subjects': {'n_subjects_per_shard': 1000}},
         'convert_to_subject_sharded',
         {'convert_to_MEDS_events': {'do_dedup_text_and_numeric': False}}]

        Unknown keys are rejected by name, listing the allowed set; options are
        type-validated against their stage's contract (bools are rejected where
        ints are expected — YAML ``true`` must not pass as 1); version/name values
        must be strings (an unquoted YAML ``3.1`` parses as a float and gets a
        targeted quote-it message):

        >>> EtlConfig.parse({"raw_dataset_version": "1", "pipeline": ["convert_to_parquet"]})
        Traceback (most recent call last):
            ...
        ValueError: etl: block contains unknown key(s) ['pipeline']. Allowed keys:
        ['dataset_name', 'description_separator', 'do_dedup_text_and_numeric',
        'external_splits_json_fp', 'n_subjects_per_shard', 'raw_dataset_version',
        'split_fracs']. The stage sequence itself is not configurable here —
        nonstandard pipeline shapes are served by writing a custom pipeline YAML and running
        `MEDS_transform-pipeline` directly (see the README's "Custom pipeline shapes").
        >>> EtlConfig.parse({"n_subjects_per_shard": True})
        Traceback (most recent call last):
            ...
        ValueError: etl.n_subjects_per_shard (a `split_and_shard_subjects` option) must be a
        positive int, got bool (True).
        >>> EtlConfig(raw_dataset_version=3.1)
        Traceback (most recent call last):
            ...
        ValueError: etl.raw_dataset_version must be a non-empty string, got float (3.1).
        Quote the version in YAML: raw_dataset_version: "3.1".
    """

    # The canonical MEDS-extraction stage sequence — the ONLY pipeline
    # `meds-extract-run` runs. The *order* is irreducible knowledge held here; the
    # *names* are cross-checked at import time against the stages this distribution
    # actually registers in the `MEDS_transforms.stages` entry-point group (see the
    # module-level check below), so a stage rename breaks loudly at import instead
    # of at run time.
    DEFAULT_PIPELINE: ClassVar[tuple[str, ...]] = (
        "convert_to_parquet",
        "split_and_shard_subjects",
        "convert_to_subject_sharded",
        "convert_to_MEDS_events",
        "extract_code_metadata",
        "merge_to_MEDS_cohort",
        "finalize_MEDS_metadata",
        "finalize_MEDS_data",
    )

    # The curated per-stage options an ``etl:`` block may set:
    # option -> (owning stage, value validator). Names are the real
    # stage-parameter names, verbatim — no aliases.
    _STAGE_OPTIONS: ClassVar[dict[str, tuple[str, Callable[[Any], str | None]]]] = {
        "n_subjects_per_shard": ("split_and_shard_subjects", positive_int_err),
        "split_fracs": ("split_and_shard_subjects", nonempty_mapping_err),
        "external_splits_json_fp": ("split_and_shard_subjects", nonempty_str_err),
        "do_dedup_text_and_numeric": ("convert_to_MEDS_events", bool_err),
        "description_separator": ("extract_code_metadata", nonempty_str_err),
    }

    # The complete key set an ``etl:`` block may carry; the unknown-key error
    # derives its allowed list from this so the two can't drift.
    ALLOWED_KEYS: ClassVar[frozenset[str]] = frozenset(
        {"dataset_name", "raw_dataset_version", *_STAGE_OPTIONS}
    )

    dataset_name: str | None = None
    raw_dataset_version: str | None = None
    stage_options: dict[str, Any] = field(default_factory=dict)

    def __post_init__(self):
        if self.raw_dataset_version is not None and (err := nonempty_str_err(self.raw_dataset_version)):
            raise ValueError(
                f"etl.raw_dataset_version {err}. Quote the version in YAML: "
                f'raw_dataset_version: "{self.raw_dataset_version}".'
            )
        if self.dataset_name is not None and (err := nonempty_str_err(self.dataset_name)):
            raise ValueError(f"etl.dataset_name {err} (omitting it is also fine).")
        for opt, value in self.stage_options.items():
            if opt not in self._STAGE_OPTIONS:
                raise ValueError(f"Unknown etl stage option {opt!r}. Allowed: {sorted(self._STAGE_OPTIONS)}.")
            stage, validator = self._STAGE_OPTIONS[opt]
            if err := validator(value):
                raise ValueError(f"etl.{opt} (a `{stage}` option) {err}.")

    @classmethod
    def parse(cls, raw: Mapping[str, Any] | DictConfig | None) -> EtlConfig:
        """Parse and validate a raw ``etl:`` mapping (``None`` => all defaults)."""
        if OmegaConf.is_config(raw):
            raw = OmegaConf.to_container(raw, resolve=False)
        if raw is None:
            raw = {}
        if not isinstance(raw, Mapping):
            raise ValueError(
                f"etl: block must be a mapping with keys among {sorted(cls.ALLOWED_KEYS)}, "
                f"got {type(raw).__name__}."
            )
        unknown = sorted(set(raw) - cls.ALLOWED_KEYS)
        if unknown:
            raise ValueError(
                f"etl: block contains unknown key(s) {unknown}. Allowed keys: "
                f"{sorted(cls.ALLOWED_KEYS)}. The stage sequence itself is not configurable "
                f"here — nonstandard pipeline shapes are served by writing a custom pipeline "
                f"YAML and running `MEDS_transform-pipeline` directly (see the README's "
                f'"Custom pipeline shapes").'
            )
        return cls(
            dataset_name=raw.get("dataset_name"),
            raw_dataset_version=raw.get("raw_dataset_version"),
            stage_options={k: raw[k] for k in cls._STAGE_OPTIONS if k in raw},
        )

    def stages_container(self) -> list[str | dict]:
        """The ``stages:`` list for a MEDS-transforms pipeline config, as plain containers.

        The canonical :attr:`DEFAULT_PIPELINE` sequence, with each curated stage
        option attached to its owning stage (stages with no options stay bare names).
        """
        by_stage: dict[str, dict[str, Any]] = {}
        for opt, value in self.stage_options.items():
            stage = self._STAGE_OPTIONS[opt][0]
            by_stage.setdefault(stage, {})[opt] = value
        return [{stage: by_stage[stage]} if stage in by_stage else stage for stage in self.DEFAULT_PIPELINE]

parse(raw) classmethod

Parse and validate a raw etl: mapping (None => all defaults).

Source code in MEDS_extract/config.py

@classmethod
def parse(cls, raw: Mapping[str, Any] | DictConfig | None) -> EtlConfig:
    """Parse and validate a raw ``etl:`` mapping (``None`` => all defaults)."""
    if OmegaConf.is_config(raw):
        raw = OmegaConf.to_container(raw, resolve=False)
    if raw is None:
        raw = {}
    if not isinstance(raw, Mapping):
        raise ValueError(
            f"etl: block must be a mapping with keys among {sorted(cls.ALLOWED_KEYS)}, "
            f"got {type(raw).__name__}."
        )
    unknown = sorted(set(raw) - cls.ALLOWED_KEYS)
    if unknown:
        raise ValueError(
            f"etl: block contains unknown key(s) {unknown}. Allowed keys: "
            f"{sorted(cls.ALLOWED_KEYS)}. The stage sequence itself is not configurable "
            f"here — nonstandard pipeline shapes are served by writing a custom pipeline "
            f"YAML and running `MEDS_transform-pipeline` directly (see the README's "
            f'"Custom pipeline shapes").'
        )
    return cls(
        dataset_name=raw.get("dataset_name"),
        raw_dataset_version=raw.get("raw_dataset_version"),
        stage_options={k: raw[k] for k in cls._STAGE_OPTIONS if k in raw},
    )

stages_container()

The stages: list for a MEDS-transforms pipeline config, as plain containers.

The canonical :attr:DEFAULT_PIPELINE sequence, with each curated stage option attached to its owning stage (stages with no options stay bare names).

Source code in MEDS_extract/config.py

def stages_container(self) -> list[str | dict]:
    """The ``stages:`` list for a MEDS-transforms pipeline config, as plain containers.

    The canonical :attr:`DEFAULT_PIPELINE` sequence, with each curated stage
    option attached to its owning stage (stages with no options stay bare names).
    """
    by_stage: dict[str, dict[str, Any]] = {}
    for opt, value in self.stage_options.items():
        stage = self._STAGE_OPTIONS[opt][0]
        by_stage.setdefault(stage, {})[opt] = value
    return [{stage: by_stage[stage]} if stage in by_stage else stage for stage in self.DEFAULT_PIPELINE]

EventConfig dataclass

A single MEDS event extraction config.

columns maps each output column name to its parsed dftly node. "code" is mandatory; "time" is optional and may be None (static event). Every non-None value is a :class:dftly.nodes.base.NodeBase instance — :meth:parse handles converting raw input (strings or expanded dftly dicts) into nodes.

metadata holds the raw _metadata block, or {} if absent. It’s consumed separately by extract_code_metadata, but every per-prefix entry is compiled and validated here, at construction time (via :func:compile_metadata_block), so config mistakes — a _metadata block on a literal code, a block producing no join-key columns, a reserved output name — surface when the MESSY file is loaded, in every stage, rather than mid-pipeline.

Examples:

>>> from dftly import Parser
>>> p = Parser()
>>> ev = EventConfig(
...     name="lab",
...     columns={"code": p('f"{$test}//{$units}"'),
...              "time": p('$ts::"%Y-%m-%d"'),
...              "numeric_value": p("$result")},
... )
>>> ev.is_static
False
>>> sorted(ev.referenced_columns)
['result', 'test', 'ts', 'units']
>>> static = EventConfig(name="eye_color", columns={"code": p('"EYE_COLOR"'), "time": None})
>>> static.is_static
True

Direct construction validates the same invariants as :meth:parse: missing code, per-event subject_id, and non-node column values all raise on instantiation.

>>> EventConfig(name="bad", columns={"time": None})
Traceback (most recent call last):
    ...
KeyError: "Event 'bad' must contain a 'code' key. Got: [time]."
>>> EventConfig(name="bad", columns={"code": p("X"), "subject_id": p("$sid")})
Traceback (most recent call last):
    ...
ValueError: Event 'bad' contains a 'subject_id' key. subject_id is a table-level concept ...
>>> EventConfig(name="bad", columns={"code": "X"})
Traceback (most recent call last):
    ...
TypeError: Event 'bad' column 'code' must be a parsed dftly node, got str ('X').

_metadata blocks are validated at construction too — e.g. a block whose produced columns include no code component is rejected here, not mid-stage:

>>> EventConfig(
...     name="chart",
...     columns={"code": p('f"CHART//{$itemid}"'), "time": None},
...     metadata={"d_items": {"description": "$label"}},
...     raw_code='f"CHART//{$itemid}"',
... )
Traceback (most recent call last):
    ...
ValueError: _metadata block (event 'chart', metadata prefix 'd_items') produces no
join-key columns: ...

A metadata-carrying event must retain its code expression’s raw source string (it becomes the code_template provenance column; a parsed node has no faithful string rendering), so raw_code is required alongside metadata:

>>> EventConfig(
...     name="chart",
...     columns={"code": p('f"CHART//{$itemid}"'), "time": None},
...     metadata={"d_items": {"itemid": "$itemid", "description": "$label"}},
... )
Traceback (most recent call last):
    ...
ValueError: Event 'chart' declares a _metadata block but its 'code' was not given as a
dftly expression string. ...

Source code in MEDS_extract/config.py

@dataclass(frozen=True)
class EventConfig:
    """A single MEDS event extraction config.

    ``columns`` maps each output column name to its parsed dftly node. ``"code"``
    is mandatory; ``"time"`` is optional and may be ``None`` (static event).
    Every non-None value is a :class:`dftly.nodes.base.NodeBase` instance —
    :meth:`parse` handles converting raw input (strings or expanded dftly
    dicts) into nodes.

    ``metadata`` holds the raw ``_metadata`` block, or ``{}`` if absent. It's
    consumed separately by ``extract_code_metadata``, but every per-prefix entry
    is compiled and validated **here, at construction time** (via
    :func:`compile_metadata_block`), so config mistakes — a ``_metadata`` block
    on a literal code, a block producing no join-key columns, a reserved output
    name — surface when the MESSY file is loaded, in
    every stage, rather than mid-pipeline.

    Examples:
        >>> from dftly import Parser
        >>> p = Parser()
        >>> ev = EventConfig(
        ...     name="lab",
        ...     columns={"code": p('f"{$test}//{$units}"'),
        ...              "time": p('$ts::"%Y-%m-%d"'),
        ...              "numeric_value": p("$result")},
        ... )
        >>> ev.is_static
        False
        >>> sorted(ev.referenced_columns)
        ['result', 'test', 'ts', 'units']
        >>> static = EventConfig(name="eye_color", columns={"code": p('"EYE_COLOR"'), "time": None})
        >>> static.is_static
        True

        Direct construction validates the same invariants as :meth:`parse`:
        missing ``code``, per-event ``subject_id``, and non-node column values
        all raise on instantiation.

        >>> EventConfig(name="bad", columns={"time": None})
        Traceback (most recent call last):
            ...
        KeyError: "Event 'bad' must contain a 'code' key. Got: [time]."
        >>> EventConfig(name="bad", columns={"code": p("X"), "subject_id": p("$sid")})
        Traceback (most recent call last):
            ...
        ValueError: Event 'bad' contains a 'subject_id' key. subject_id is a table-level concept ...
        >>> EventConfig(name="bad", columns={"code": "X"})
        Traceback (most recent call last):
            ...
        TypeError: Event 'bad' column 'code' must be a parsed dftly node, got str ('X').

        ``_metadata`` blocks are validated at construction too — e.g. a block whose
        produced columns include no code component is rejected here, not mid-stage:

        >>> EventConfig(
        ...     name="chart",
        ...     columns={"code": p('f"CHART//{$itemid}"'), "time": None},
        ...     metadata={"d_items": {"description": "$label"}},
        ...     raw_code='f"CHART//{$itemid}"',
        ... )
        Traceback (most recent call last):
            ...
        ValueError: _metadata block (event 'chart', metadata prefix 'd_items') produces no
        join-key columns: ...

        A metadata-carrying event must retain its code expression's raw source string
        (it becomes the ``code_template`` provenance column; a parsed node has no
        faithful string rendering), so ``raw_code`` is required alongside ``metadata``:

        >>> EventConfig(
        ...     name="chart",
        ...     columns={"code": p('f"CHART//{$itemid}"'), "time": None},
        ...     metadata={"d_items": {"itemid": "$itemid", "description": "$label"}},
        ... )
        Traceback (most recent call last):
            ...
        ValueError: Event 'chart' declares a _metadata block but its 'code' was not given as a
        dftly expression string. ...
    """

    name: str
    columns: dict[str, NodeBase | None]
    metadata: dict = field(default_factory=dict)
    raw_code: str | None = None

    def __post_init__(self):
        if "code" not in self.columns:
            raise KeyError(
                f"Event '{self.name}' must contain a 'code' key. Got: [{', '.join(self.columns.keys())}]."
            )
        if "subject_id" in self.columns:
            raise ValueError(
                f"Event '{self.name}' contains a 'subject_id' key. subject_id is a table-level "
                f"concept and must be set in '_defaults', not per-event."
            )
        for k, v in self.columns.items():
            if k == "time" and v is None:
                continue
            if not isinstance(v, NodeBase):
                raise TypeError(
                    f"Event '{self.name}' column '{k}' must be a parsed dftly node, "
                    f"got {type(v).__name__} ({v!r})."
                )
        # Validate every ``_metadata`` prefix block at construction so config mistakes
        # surface at MESSY-load time in every stage (repo validation-at-construction
        # convention). The compiled result is discarded here; ``extract_code_metadata``
        # recompiles through the same function when it runs.
        if self.metadata:
            # ``extract_code_metadata`` stamps the code expression's raw dftly SOURCE
            # STRING on every extracted metadata row as ``code_template`` (a mandated,
            # human-readable provenance column). A pre-parsed node has no faithful
            # string rendering (``repr(node)`` is not dftly), so a metadata-carrying
            # event must have its code written as a string expression.
            if self.raw_code is None:
                raise ValueError(
                    f"Event '{self.name}' declares a _metadata block but its 'code' was not given "
                    f"as a dftly expression string. Metadata extraction stamps the raw code string "
                    f"on its outputs as 'code_template', so write 'code' as a string expression "
                    f"(or pass raw_code= when constructing EventConfig directly)."
                )
            for prefix, block in self.metadata.items():
                compile_fn = (
                    compile_self_metadata_block if prefix == SELF_METADATA_PREFIX else compile_metadata_block
                )
                compile_fn(
                    block,
                    self.code_source_columns,
                    code_template_str=self.raw_code,
                    context=f"event '{self.name}', metadata prefix '{prefix}'",
                )

    @classmethod
    def parse(cls, name: str, raw: Mapping[str, Any]) -> EventConfig:
        """Parse a raw event block into an EventConfig.

        Each column value is compiled through :class:`dftly.Parser`, so raw
        input may be either a dftly expression string (``"$col"``,
        ``'f"PREFIX//{$col}"'``, ``"hash($col)"``) or an expanded dftly dict
        form. The time column is the only key that may be ``None`` — a
        ``None`` time produces a static event.

        Examples:
            Strings get parsed to nodes:

            >>> ev = EventConfig.parse("lab", {"code": "X", "time": None, "numeric_value": "$v"})
            >>> type(ev.columns["code"]).__name__
            'Literal'
            >>> type(ev.columns["numeric_value"]).__name__
            'Column'
            >>> ev.columns["time"] is None
            True

            Validation errors surface at parse time with the event name:

            >>> EventConfig.parse("bad", {"time": None})
            Traceback (most recent call last):
                ...
            KeyError: "Event 'bad' must contain a 'code' key. Got: [time]."
            >>> EventConfig.parse("bad", {"code": "X", "subject_id": "$sid"})
            Traceback (most recent call last):
                ...
            ValueError: Event 'bad' contains a 'subject_id' key. subject_id is a table-level concept ...

            An underscore-prefixed key other than ``_metadata`` is almost certainly a
            typo of that reserved name, and is rejected up front rather than falling
            through to column parsing with a misleading error:

            >>> EventConfig.parse("dob", {
            ...     "code": "BIRTH",
            ...     "time": "$dob",
            ...     "_metdata": {"d_items": {"itemid": "$itemid", "description": "$label"}},
            ... })
            Traceback (most recent call last):
                ...
            ValueError: Event 'dob' has unknown reserved key(s) ['_metdata']. The only reserved
            key at the event level is '_metadata'. Output column names may not begin with an
            underscore.

            ``_metadata`` blocks are compiled and validated here too (via
            :func:`compile_metadata_block`, which documents the full error catalog), so
            every config mistake fires at parse time with the event and prefix named.
            A block on a literal code:

            >>> EventConfig.parse("admit", {
            ...     "code": "ADMISSION",
            ...     "time": None,
            ...     "_metadata": {"adm_meta": {"description": "$title"}},
            ... })
            Traceback (most recent call last):
                ...
            ValueError: The code expression 'ADMISSION' is a literal: ... no components to match
            metadata on. ...

            A block that produces no join key, naming the components the event offers:

            >>> EventConfig.parse("med", {
            ...     "code": 'f"{$medication_name}//{$dose}"',
            ...     "time": None,
            ...     "_metadata": {"med_classes": {"description": "$drug_class"}},
            ... })
            Traceback (most recent call last):
                ...
            ValueError: _metadata block (event 'med', metadata prefix 'med_classes') produces no
            join-key columns: ... Component columns available on this event: ['dose',
            'medication_name'] ...

            A reserved output name:

            >>> EventConfig.parse("chart", {
            ...     "code": 'f"CHART//{$itemid}"',
            ...     "time": None,
            ...     "_metadata": {"d_items": {"itemid": "$itemid", "code_template": "$label"}},
            ... })
            Traceback (most recent call last):
                ...
            ValueError: _metadata output column name(s) ['code_template'] are reserved: ...

            A value that is not a dftly expression, named with its event and prefix:

            >>> EventConfig.parse("lab", {
            ...     "code": "$test_name",
            ...     "time": None,
            ...     "_metadata": {
            ...         "lab_meta": {"test_name": "$test_name", "description": ["special_title", "title"]}
            ...     },
            ... })
            Traceback (most recent call last):
                ...
            ValueError: _metadata column 'description' (event 'lab', metadata prefix 'lab_meta')
            failed to parse as a dftly expression: ...

            And the one nuance on reserved names: a key named ``code`` is legal exactly
            when the code expression references a source column literally named ``code``
            (the ICD/OMOP vocabulary shape) — it is a join key, never an output:

            >>> ev = EventConfig.parse("dx", {
            ...     "code": 'f"ICD//{$code}"',
            ...     "time": None,
            ...     "_metadata": {"icd_meta": {"code": "$code", "description": "$long_title"}},
            ... })
            >>> ev.metadata
            {'icd_meta': {'code': '$code', 'description': '$long_title'}}
        """
        raw = dict(raw)
        metadata = dict(raw.pop("_metadata", {}))

        stray_reserved = sorted(k for k in raw if k.startswith("_"))
        if stray_reserved:
            raise ValueError(
                f"Event '{name}' has unknown reserved key(s) {stray_reserved}. The only reserved "
                f"key at the event level is '_metadata'. Output column names may not begin with "
                f"an underscore."
            )

        raw_code = raw["code"] if isinstance(raw.get("code"), str) else None

        columns: dict[str, NodeBase | None] = {}
        parser = Parser()
        for k, v in raw.items():
            if k == "time" and v is None:
                columns[k] = None
                continue
            try:
                columns[k] = v if isinstance(v, NodeBase) else parser(v)
            except Exception as e:
                raise ValueError(f"Event '{name}' column '{k}' failed to parse: {e}") from e

        return cls(name=name, columns=columns, metadata=metadata, raw_code=raw_code)

    @property
    def is_static(self) -> bool:
        """True if this event has no time column (time absent or ``None``)."""
        return self.columns.get("time") is None

    @cached_property
    def polars_exprs(self) -> dict[str, pl.Expr]:
        """Polars expression for each output column. Built once and reused.

        ``"time"`` resolves to a typed null literal for static events.
        """
        out: dict[str, pl.Expr] = {}
        for k, v in self.columns.items():
            if k == "time" and v is None:
                out[k] = pl.lit(None, dtype=pl.Datetime)
            else:
                out[k] = v.polars_expr
        return out

    @cached_property
    def code_source_columns(self) -> frozenset[str]:
        """Source columns referenced by the ``code`` expression."""
        return frozenset(self.columns["code"].referenced_columns)

    @cached_property
    def self_metadata_exprs(self) -> dict[str, NodeBase]:
        """Compiled ``_self`` metadata expressions (output column → node), or ``{}``.

        ``_self`` metadata is evaluated over the event's own prepared source frame
        during :meth:`extract` — the outputs land in the ``METADATA_COMPONENTS_COL``
        struct rather than re-reading any raw table at metadata-extraction time.
        """
        block = self.metadata.get(SELF_METADATA_PREFIX)
        if not block:
            return {}
        compiled = compile_self_metadata_block(
            block,
            self.code_source_columns,
            code_template_str=self.raw_code,
            context=f"event '{self.name}', metadata prefix '{SELF_METADATA_PREFIX}'",
        )
        return compiled.exprs

    @cached_property
    def referenced_columns(self) -> frozenset[str]:
        """All source columns referenced by any output column expression.

        Aggregates across ``code``, ``time``, all additional value columns, and any
        ``_self`` metadata expressions (which read the same prepared frame — external
        ``_metadata`` blocks reference metadata-table columns and are excluded). Used
        by :meth:`TableConfig.source_columns` to determine which columns must be read
        from the source parquet file.
        """
        cols: set[str] = set()
        for v in self.columns.values():
            if v is None:
                continue
            cols.update(v.referenced_columns)
        for node in self.self_metadata_exprs.values():
            cols.update(node.referenced_columns)
        return frozenset(cols)

    def extract(
        self,
        df: pl.LazyFrame,
        source_block: str,
        do_dedup_text_and_numeric: bool = False,
    ) -> pl.LazyFrame:
        """Extract this event's rows from a dataframe already prepared by :meth:`TableConfig.prepare`.

        The input ``df`` must have a ``subject_id`` column and any source columns this
        event references. ``source_block`` tags each output row with its MESSY origin
        (e.g. ``"patients/eye_color"``) and is always included in the output schema.

        **The output is de-duplicated.** Two source rows that produce byte-identical
        event rows (every output column equal, including ``code_components``) collapse
        into one. Genuine repeated measurements are unaffected as long as *something*
        distinguishes them — a differing time, value, or code component — but a source
        table that legitimately records the same value twice at the same timestamp, with
        no distinguishing column extracted, yields one event, not two. This is intended:
        the same raw row reaching extraction twice (a re-run, an overlapping shard, a
        fan-out from a non-unique join target) must not inflate the cohort.

        Examples:
            >>> _ = pl.Config.set_tbl_width_chars(600)
            >>> raw = pl.DataFrame({
            ...     "subject_id": [1, 2, 3],
            ...     "color": ["blue", "green", "brown"],
            ... })
            >>> ev = EventConfig.parse(
            ...     "eye_color",
            ...     {"code": "EYE_COLOR", "time": None, "eye_color": "$color"},
            ... )
            >>> ev.extract(raw.lazy(), "patients/eye_color").collect()
            shape: (3, 5)
            ┌────────────┬───────────┬──────────────┬───────────┬────────────────────┐
            │ subject_id ┆ code      ┆ time         ┆ eye_color ┆ source_block       │
            │ ---        ┆ ---       ┆ ---          ┆ ---       ┆ ---                │
            │ i64        ┆ str       ┆ datetime[μs] ┆ str       ┆ str                │
            ╞════════════╪═══════════╪══════════════╪═══════════╪════════════════════╡
            │ 1          ┆ EYE_COLOR ┆ null         ┆ blue      ┆ patients/eye_color │
            │ 2          ┆ EYE_COLOR ┆ null         ┆ green     ┆ patients/eye_color │
            │ 3          ┆ EYE_COLOR ┆ null         ┆ brown     ┆ patients/eye_color │
            └────────────┴───────────┴──────────────┴───────────┴────────────────────┘

            A row whose computed ``time`` is null, or whose **bare-column** ``code``
            (``code: $col``) is null, is dropped after selection (with a per-event
            WARNING summarizing the drop counts):

            >>> raw = pl.DataFrame({
            ...     "subject_id": [1, 2, 3],
            ...     "name": ["A", None, "C"],
            ...     "ts": ["2021-01-01", "2021-01-02", None],
            ... })
            >>> ev = EventConfig.parse("e", {"code": "$name", "time": '$ts::"%Y-%m-%d"'})
            >>> ev.extract(raw.lazy(), "t/e").collect().select("subject_id", "code", "time")
            shape: (1, 3)
            ┌────────────┬──────┬────────────┐
            │ subject_id ┆ code ┆ time       │
            │ ---        ┆ ---  ┆ ---        │
            │ i64        ┆ str  ┆ date       │
            ╞════════════╪══════╪════════════╡
            │ 1          ┆ A    ┆ 2021-01-01 │
            └────────────┴──────┴────────────┘

            A single time column mixing several formats is handled by coalescing lenient
            (``::?``) parses — each row takes the first format that matches, and rows
            matching none (garbage, ``""``, null) get a null time and are dropped (and
            counted in the WARNING). This is the documented multi-format idiom; a strict
            (``::"fmt"``) cast would instead *error* on the first unparsable value:

            >>> raw = pl.DataFrame({
            ...     "subject_id": [1, 2, 3, 4],
            ...     "ts": ["01/02/21 12:30:00", "01/03/21", "not a date", None],
            ... })
            >>> ev = EventConfig.parse("visit", {
            ...     "code": "VISIT",
            ...     "time": 'coalesce($ts::?"%m/%d/%y %H:%M:%S", $ts::?"%m/%d/%y")',
            ... })
            >>> ev.extract(raw.lazy(), "visits/visit").collect().select("subject_id", "time")
            shape: (2, 2)
            ┌────────────┬─────────────────────┐
            │ subject_id ┆ time                │
            │ ---        ┆ ---                 │
            │ i64        ┆ datetime[μs]        │
            ╞════════════╪═════════════════════╡
            │ 1          ┆ 2021-01-02 12:30:00 │
            │ 2          ┆ 2021-01-03 00:00:00 │
            └────────────┴─────────────────────┘

            A *composite / interpolated* code null-propagates the same way: if **any** referenced
            component is null, the whole code is null, so the row is dropped. Below, only the row
            with both ``$itemid`` and ``$valueuom`` present survives:

            >>> raw = pl.DataFrame({
            ...     "subject_id": [1, 2, 3, 4],
            ...     "itemid": ["GLU", "GLU", None, None],  # present, present, null, null
            ...     "valueuom": ["mg/dL", None, "mg/dL", None],  # present, null, present, null
            ... })
            >>> ev = EventConfig.parse("lab", {"code": 'f"{$itemid}//{$valueuom}"', "time": None})
            >>> ev.extract(raw.lazy(), "labs/lab").collect().select("subject_id", "code")
            shape: (1, 2)
            ┌────────────┬────────────┐
            │ subject_id ┆ code       │
            │ ---        ┆ ---        │
            │ i64        ┆ str        │
            ╞════════════╪════════════╡
            │ 1          ┆ GLU//mg/dL │
            └────────────┴────────────┘

            To keep rows with missing components, coalesce them in the MESSY expression with the
            dftly ``??`` operator (single-quote the literal fallback inside the f-string) — each
            null component is filled with your chosen literal, so all four rows are retained:

            >>> ev = EventConfig.parse(
            ...     "lab",
            ...     {"code": '''f"{$itemid ?? 'UNK'}//{$valueuom ?? 'UNK'}"''', "time": None},
            ... )
            >>> ev.extract(raw.lazy(), "labs/lab").collect().sort("subject_id").select(
            ...     "subject_id", "code"
            ... )
            shape: (4, 2)
            ┌────────────┬────────────┐
            │ subject_id ┆ code       │
            │ ---        ┆ ---        │
            │ i64        ┆ str        │
            ╞════════════╪════════════╡
            │ 1          ┆ GLU//mg/dL │
            │ 2          ┆ GLU//UNK   │
            │ 3          ┆ UNK//mg/dL │
            │ 4          ┆ UNK//UNK   │
            └────────────┴────────────┘

            The fallback is per-component and any literal you choose — coalescing only some
            components, or with distinct markers, are both fine (see the README's *Code
            Construction* section for more variations).

            With ``do_dedup_text_and_numeric=True``, a ``text_value`` that
            numerically equals ``numeric_value`` is nulled out:

            >>> raw = pl.DataFrame({
            ...     "subject_id": [1, 2],
            ...     "ts": ["2021-01-01", "2021-01-02"],
            ...     "val": [1.5, 2.0],
            ...     "text": ["1.5", "other"],
            ... })
            >>> ev = EventConfig.parse("m", {
            ...     "code": "MEAS",
            ...     "time": '$ts::"%Y-%m-%d"',
            ...     "numeric_value": "$val",
            ...     "text_value": "$text",
            ... })
            >>> ev.extract(raw.lazy(), "t/m", do_dedup_text_and_numeric=True).collect().select(
            ...     "numeric_value", "text_value"
            ... )
            shape: (2, 2)
            ┌───────────────┬────────────┐
            │ numeric_value ┆ text_value │
            │ ---           ┆ ---        │
            │ f64           ┆ str        │
            ╞═══════════════╪════════════╡
            │ 1.5           ┆ null       │
            │ 2.0           ┆ other      │
            └───────────────┴────────────┘

            The ``code_components`` struct's fields are named for the *source* columns the
            code references — including when a source column is literally named ``code``
            (the idiomatic ICD/OMOP vocabulary shape; ``extract_code_metadata`` unnests this
            struct and must alias the assembled code away from it):

            >>> raw = pl.DataFrame({"subject_id": [1], "code": ["250.00"], "ts": ["2020-01-01"]})
            >>> ev = EventConfig.parse("dx", {"code": 'f"ICD//{$code}"', "time": '$ts::"%Y-%m-%d"'})
            >>> ev.extract(raw.lazy(), "diagnoses/dx").collect().schema["code_components"]
            Struct({'code': String})

            A non-static event whose ``time`` expression is not temporally typed is rejected
            here, per event block — downstream schema alignment would otherwise silently
            reinterpret integers as 1970-epoch offsets, and strings would sort
            lexicographically (#207). Both the raw-integer-offset and the forgotten-cast
            string shapes are caught:

            >>> raw = pl.DataFrame({"subject_id": [1], "hr": [80.0], "offset": [1444]})
            >>> ev = EventConfig.parse("hr", {"code": "HR", "time": "$offset", "numeric_value": "$hr"})
            >>> ev.extract(raw.lazy(), "vitals/hr")
            Traceback (most recent call last):
                ...
            ValueError: `vitals/hr`: the `time` expression produced dtype Int64, not a date/datetime. ...
            >>> raw = pl.DataFrame({"subject_id": [1], "code": ["250.00"], "ts": ["2020-01-01"]})
            >>> ev = EventConfig.parse("dx", {"code": 'f"ICD//{$code}"', "time": "$ts"})
            >>> ev.extract(raw.lazy(), "diagnoses/dx")
            Traceback (most recent call last):
                ...
            ValueError: `diagnoses/dx`: the `time` expression produced dtype String, not a date/datetime. ...
        """
        exprs: dict[str, pl.Expr] = {"subject_id": pl.col("subject_id")}

        # `code` is the native dftly expression. String interpolation null-propagates: if any
        # referenced component is null the whole code is null, and the row is dropped below (a MEDS
        # code may not be null). Authors opt into retaining such rows by coalescing components in
        # the MESSY expression, e.g. `f"{$itemid ?? 'UNK'}//{$valueuom ?? 'UNK'}"`.
        exprs["code"] = self.polars_exprs["code"]
        if self.code_source_columns:
            # Raw, typed component values (the `code` string above is derived from these).
            exprs["code_components"] = pl.struct(
                **{col: pl.col(col) for col in sorted(self.code_source_columns)}
            )
        if self.self_metadata_exprs:
            # Evaluated ``_self`` metadata outputs, paired row-wise with the components
            # above. ``extract_code_metadata`` reads distinct component→metadata pairs
            # from these two structs; ``merge_to_MEDS_cohort`` drops both.
            exprs[METADATA_COMPONENTS_COL] = pl.struct(
                **{out: node.polars_expr for out, node in self.self_metadata_exprs.items()}
            )

        exprs["time"] = self.polars_exprs["time"]

        for k in self.columns:
            if k in ("code", "time"):
                continue
            exprs[k] = self.polars_exprs[k]

        if do_dedup_text_and_numeric and "numeric_value" in exprs and "text_value" in exprs:
            text_expr = exprs["text_value"]
            num_expr = exprs["numeric_value"]
            exprs["text_value"] = (
                pl.when(text_expr.cast(pl.Float32, strict=False) == num_expr.cast(pl.Float32))
                .then(pl.lit(None, pl.String))
                .otherwise(text_expr)
            )

        exprs[SOURCE_BLOCK_COL] = pl.lit(source_block)

        out = df.select(**exprs)

        # A non-static `time` must be temporally typed HERE, per event block, because nothing
        # downstream can catch the mistake intelligibly: DataSchema.align at finalize
        # reinterprets Int64 as 1970-epoch microseconds with no error (#207), merge's
        # diagonal_relaxed concat degrades sibling shards' correct Datetime columns to match a
        # wrong one, and a String time sorts lexicographically. Null passes: an all-null source
        # column resolves to dtype Null, and those rows are dropped (and counted) just below.
        if not self.is_static:
            time_dtype = out.collect_schema()["time"]
            if time_dtype != pl.Null and not isinstance(time_dtype, pl.Datetime | pl.Date):
                raise ValueError(
                    f"`{source_block}`: the `time` expression produced dtype {time_dtype}, not a "
                    "date/datetime. Integer offset columns must be converted to timestamps (e.g. a "
                    "`_table.cols` pseudotime chain adding the offset to an anchor time), and string "
                    'columns need an explicit cast (`$col::"%Y-%m-%d %H:%M:%S"`-style).'
                )

        # Null-`code` / null-`time` rows are dropped by filtering the COMPUTED columns, after the
        # select — never via a fallible predicate over the raw source expressions. Polars pushes
        # eligible predicates into the parquet scan itself, where they are evaluated on
        # validity-unmasked data (null slots of a String column surface as ""), so a predicate
        # containing a strict cast/strptime panics on polars >= 1.28 (and mis-evaluates silently
        # before that): https://github.com/pola-rs/polars/issues/28521. A predicate on a computed
        # column instead stays above the SELECT boundary (verified on polars 1.26-1.43 and
        # regression-guarded in tests), and the scan's projection pushdown is preserved. Note the
        # deliberate consequence for strict (`::"fmt"`) time casts: an unparsable non-null value
        # (e.g. "") now raises an InvalidOperationError naming the value, instead of being
        # silently pre-filtered; authors who want unparsable values dropped opt in with a
        # lenient (`::?"fmt"`) cast, which nulls them so they are dropped (and counted) here.
        drop_null_code = bool(self.code_source_columns)
        drop_null_time = not self.is_static

        if drop_null_code or drop_null_time:
            # Drop accounting: one aggregation pass over the pre-filter frame so silently
            # vanishing rows (unparsable times, null code components) are surfaced per event.
            stats_exprs = [pl.len().alias("n_total")]
            if drop_null_code:
                stats_exprs.append(pl.col("code").is_null().sum().alias("n_null_code"))
            if drop_null_time:
                stats_exprs.append(pl.col("time").is_null().sum().alias("n_null_time"))
            stats = out.select(stats_exprs).collect()
            n_total = stats["n_total"][0]
            n_null_code = stats["n_null_code"][0] if drop_null_code else 0
            n_null_time = stats["n_null_time"][0] if drop_null_time else 0
            if n_null_time or n_null_code:
                parts = []
                if n_null_time:
                    parts.append(
                        f"{n_null_time}/{n_total} rows with null time "
                        f"(unparsable or missing under the configured formats)"
                    )
                if n_null_code:
                    count = f"{n_null_code}" if n_null_time else f"{n_null_code}/{n_total} rows"
                    parts.append(f"{count} with null code")
                logger.warning(f"`{source_block}`: dropped " + " and ".join(parts))

        if drop_null_code:
            # A MEDS `code` may never be null. A bare-column code that is null, or an interpolated
            # code with a null (un-coalesced) component, evaluates to a null code — drop those
            # rows. Rows the author rescued with a `??` coalesce never evaluate to null, so they
            # are kept.
            out = out.filter(pl.col("code").is_not_null())
        if drop_null_time:
            out = out.filter(pl.col("time").is_not_null())

        return out.unique(maintain_order=True)

code_source_columns cached property

Source columns referenced by the code expression.

is_static property

True if this event has no time column (time absent or None).

polars_exprs cached property

Polars expression for each output column. Built once and reused.

"time" resolves to a typed null literal for static events.

referenced_columns cached property

All source columns referenced by any output column expression.

Aggregates across code, time, all additional value columns, and any _self metadata expressions (which read the same prepared frame — external _metadata blocks reference metadata-table columns and are excluded). Used by :meth:TableConfig.source_columns to determine which columns must be read from the source parquet file.

self_metadata_exprs cached property

Compiled _self metadata expressions (output column → node), or {}.

_self metadata is evaluated over the event’s own prepared source frame during :meth:extract — the outputs land in the METADATA_COMPONENTS_COL struct rather than re-reading any raw table at metadata-extraction time.

extract(df, source_block, do_dedup_text_and_numeric=False)

Extract this event’s rows from a dataframe already prepared by :meth:TableConfig.prepare.

The input df must have a subject_id column and any source columns this event references. source_block tags each output row with its MESSY origin (e.g. "patients/eye_color") and is always included in the output schema.

The output is de-duplicated. Two source rows that produce byte-identical event rows (every output column equal, including code_components) collapse into one. Genuine repeated measurements are unaffected as long as something distinguishes them — a differing time, value, or code component — but a source table that legitimately records the same value twice at the same timestamp, with no distinguishing column extracted, yields one event, not two. This is intended: the same raw row reaching extraction twice (a re-run, an overlapping shard, a fan-out from a non-unique join target) must not inflate the cohort.

Examples:

>>> _ = pl.Config.set_tbl_width_chars(600)
>>> raw = pl.DataFrame({
...     "subject_id": [1, 2, 3],
...     "color": ["blue", "green", "brown"],
... })
>>> ev = EventConfig.parse(
...     "eye_color",
...     {"code": "EYE_COLOR", "time": None, "eye_color": "$color"},
... )
>>> ev.extract(raw.lazy(), "patients/eye_color").collect()
shape: (3, 5)
┌────────────┬───────────┬──────────────┬───────────┬────────────────────┐
│ subject_id ┆ code      ┆ time         ┆ eye_color ┆ source_block       │
│ ---        ┆ ---       ┆ ---          ┆ ---       ┆ ---                │
│ i64        ┆ str       ┆ datetime[μs] ┆ str       ┆ str                │
╞════════════╪═══════════╪══════════════╪═══════════╪════════════════════╡
│ 1          ┆ EYE_COLOR ┆ null         ┆ blue      ┆ patients/eye_color │
│ 2          ┆ EYE_COLOR ┆ null         ┆ green     ┆ patients/eye_color │
│ 3          ┆ EYE_COLOR ┆ null         ┆ brown     ┆ patients/eye_color │
└────────────┴───────────┴──────────────┴───────────┴────────────────────┘

A row whose computed time is null, or whose bare-column code (code: $col) is null, is dropped after selection (with a per-event WARNING summarizing the drop counts):

>>> raw = pl.DataFrame({
...     "subject_id": [1, 2, 3],
...     "name": ["A", None, "C"],
...     "ts": ["2021-01-01", "2021-01-02", None],
... })
>>> ev = EventConfig.parse("e", {"code": "$name", "time": '$ts::"%Y-%m-%d"'})
>>> ev.extract(raw.lazy(), "t/e").collect().select("subject_id", "code", "time")
shape: (1, 3)
┌────────────┬──────┬────────────┐
│ subject_id ┆ code ┆ time       │
│ ---        ┆ ---  ┆ ---        │
│ i64        ┆ str  ┆ date       │
╞════════════╪══════╪════════════╡
│ 1          ┆ A    ┆ 2021-01-01 │
└────────────┴──────┴────────────┘

A single time column mixing several formats is handled by coalescing lenient (::?) parses — each row takes the first format that matches, and rows matching none (garbage, "", null) get a null time and are dropped (and counted in the WARNING). This is the documented multi-format idiom; a strict (::"fmt") cast would instead error on the first unparsable value:

>>> raw = pl.DataFrame({
...     "subject_id": [1, 2, 3, 4],
...     "ts": ["01/02/21 12:30:00", "01/03/21", "not a date", None],
... })
>>> ev = EventConfig.parse("visit", {
...     "code": "VISIT",
...     "time": 'coalesce($ts::?"%m/%d/%y %H:%M:%S", $ts::?"%m/%d/%y")',
... })
>>> ev.extract(raw.lazy(), "visits/visit").collect().select("subject_id", "time")
shape: (2, 2)
┌────────────┬─────────────────────┐
│ subject_id ┆ time                │
│ ---        ┆ ---                 │
│ i64        ┆ datetime[μs]        │
╞════════════╪═════════════════════╡
│ 1          ┆ 2021-01-02 12:30:00 │
│ 2          ┆ 2021-01-03 00:00:00 │
└────────────┴─────────────────────┘

A composite / interpolated code null-propagates the same way: if any referenced component is null, the whole code is null, so the row is dropped. Below, only the row with both $itemid and $valueuom present survives:

>>> raw = pl.DataFrame({
...     "subject_id": [1, 2, 3, 4],
...     "itemid": ["GLU", "GLU", None, None],  # present, present, null, null
...     "valueuom": ["mg/dL", None, "mg/dL", None],  # present, null, present, null
... })
>>> ev = EventConfig.parse("lab", {"code": 'f"{$itemid}//{$valueuom}"', "time": None})
>>> ev.extract(raw.lazy(), "labs/lab").collect().select("subject_id", "code")
shape: (1, 2)
┌────────────┬────────────┐
│ subject_id ┆ code       │
│ ---        ┆ ---        │
│ i64        ┆ str        │
╞════════════╪════════════╡
│ 1          ┆ GLU//mg/dL │
└────────────┴────────────┘

To keep rows with missing components, coalesce them in the MESSY expression with the dftly ?? operator (single-quote the literal fallback inside the f-string) — each null component is filled with your chosen literal, so all four rows are retained:

>>> ev = EventConfig.parse(
...     "lab",
...     {"code": '''f"{$itemid ?? 'UNK'}//{$valueuom ?? 'UNK'}"''', "time": None},
... )
>>> ev.extract(raw.lazy(), "labs/lab").collect().sort("subject_id").select(
...     "subject_id", "code"
... )
shape: (4, 2)
┌────────────┬────────────┐
│ subject_id ┆ code       │
│ ---        ┆ ---        │
│ i64        ┆ str        │
╞════════════╪════════════╡
│ 1          ┆ GLU//mg/dL │
│ 2          ┆ GLU//UNK   │
│ 3          ┆ UNK//mg/dL │
│ 4          ┆ UNK//UNK   │
└────────────┴────────────┘

The fallback is per-component and any literal you choose — coalescing only some components, or with distinct markers, are both fine (see the README’s Code Construction section for more variations).

With do_dedup_text_and_numeric=True, a text_value that numerically equals numeric_value is nulled out:

>>> raw = pl.DataFrame({
...     "subject_id": [1, 2],
...     "ts": ["2021-01-01", "2021-01-02"],
...     "val": [1.5, 2.0],
...     "text": ["1.5", "other"],
... })
>>> ev = EventConfig.parse("m", {
...     "code": "MEAS",
...     "time": '$ts::"%Y-%m-%d"',
...     "numeric_value": "$val",
...     "text_value": "$text",
... })
>>> ev.extract(raw.lazy(), "t/m", do_dedup_text_and_numeric=True).collect().select(
...     "numeric_value", "text_value"
... )
shape: (2, 2)
┌───────────────┬────────────┐
│ numeric_value ┆ text_value │
│ ---           ┆ ---        │
│ f64           ┆ str        │
╞═══════════════╪════════════╡
│ 1.5           ┆ null       │
│ 2.0           ┆ other      │
└───────────────┴────────────┘

The code_components struct’s fields are named for the source columns the code references — including when a source column is literally named code (the idiomatic ICD/OMOP vocabulary shape; extract_code_metadata unnests this struct and must alias the assembled code away from it):

>>> raw = pl.DataFrame({"subject_id": [1], "code": ["250.00"], "ts": ["2020-01-01"]})
>>> ev = EventConfig.parse("dx", {"code": 'f"ICD//{$code}"', "time": '$ts::"%Y-%m-%d"'})
>>> ev.extract(raw.lazy(), "diagnoses/dx").collect().schema["code_components"]
Struct({'code': String})

A non-static event whose time expression is not temporally typed is rejected here, per event block — downstream schema alignment would otherwise silently reinterpret integers as 1970-epoch offsets, and strings would sort lexicographically (#207). Both the raw-integer-offset and the forgotten-cast string shapes are caught:

>>> raw = pl.DataFrame({"subject_id": [1], "hr": [80.0], "offset": [1444]})
>>> ev = EventConfig.parse("hr", {"code": "HR", "time": "$offset", "numeric_value": "$hr"})
>>> ev.extract(raw.lazy(), "vitals/hr")
Traceback (most recent call last):
    ...
ValueError: `vitals/hr`: the `time` expression produced dtype Int64, not a date/datetime. ...
>>> raw = pl.DataFrame({"subject_id": [1], "code": ["250.00"], "ts": ["2020-01-01"]})
>>> ev = EventConfig.parse("dx", {"code": 'f"ICD//{$code}"', "time": "$ts"})
>>> ev.extract(raw.lazy(), "diagnoses/dx")
Traceback (most recent call last):
    ...
ValueError: `diagnoses/dx`: the `time` expression produced dtype String, not a date/datetime. ...

Source code in MEDS_extract/config.py

def extract(
    self,
    df: pl.LazyFrame,
    source_block: str,
    do_dedup_text_and_numeric: bool = False,
) -> pl.LazyFrame:
    """Extract this event's rows from a dataframe already prepared by :meth:`TableConfig.prepare`.

    The input ``df`` must have a ``subject_id`` column and any source columns this
    event references. ``source_block`` tags each output row with its MESSY origin
    (e.g. ``"patients/eye_color"``) and is always included in the output schema.

    **The output is de-duplicated.** Two source rows that produce byte-identical
    event rows (every output column equal, including ``code_components``) collapse
    into one. Genuine repeated measurements are unaffected as long as *something*
    distinguishes them — a differing time, value, or code component — but a source
    table that legitimately records the same value twice at the same timestamp, with
    no distinguishing column extracted, yields one event, not two. This is intended:
    the same raw row reaching extraction twice (a re-run, an overlapping shard, a
    fan-out from a non-unique join target) must not inflate the cohort.

    Examples:
        >>> _ = pl.Config.set_tbl_width_chars(600)
        >>> raw = pl.DataFrame({
        ...     "subject_id": [1, 2, 3],
        ...     "color": ["blue", "green", "brown"],
        ... })
        >>> ev = EventConfig.parse(
        ...     "eye_color",
        ...     {"code": "EYE_COLOR", "time": None, "eye_color": "$color"},
        ... )
        >>> ev.extract(raw.lazy(), "patients/eye_color").collect()
        shape: (3, 5)
        ┌────────────┬───────────┬──────────────┬───────────┬────────────────────┐
        │ subject_id ┆ code      ┆ time         ┆ eye_color ┆ source_block       │
        │ ---        ┆ ---       ┆ ---          ┆ ---       ┆ ---                │
        │ i64        ┆ str       ┆ datetime[μs] ┆ str       ┆ str                │
        ╞════════════╪═══════════╪══════════════╪═══════════╪════════════════════╡
        │ 1          ┆ EYE_COLOR ┆ null         ┆ blue      ┆ patients/eye_color │
        │ 2          ┆ EYE_COLOR ┆ null         ┆ green     ┆ patients/eye_color │
        │ 3          ┆ EYE_COLOR ┆ null         ┆ brown     ┆ patients/eye_color │
        └────────────┴───────────┴──────────────┴───────────┴────────────────────┘

        A row whose computed ``time`` is null, or whose **bare-column** ``code``
        (``code: $col``) is null, is dropped after selection (with a per-event
        WARNING summarizing the drop counts):

        >>> raw = pl.DataFrame({
        ...     "subject_id": [1, 2, 3],
        ...     "name": ["A", None, "C"],
        ...     "ts": ["2021-01-01", "2021-01-02", None],
        ... })
        >>> ev = EventConfig.parse("e", {"code": "$name", "time": '$ts::"%Y-%m-%d"'})
        >>> ev.extract(raw.lazy(), "t/e").collect().select("subject_id", "code", "time")
        shape: (1, 3)
        ┌────────────┬──────┬────────────┐
        │ subject_id ┆ code ┆ time       │
        │ ---        ┆ ---  ┆ ---        │
        │ i64        ┆ str  ┆ date       │
        ╞════════════╪══════╪════════════╡
        │ 1          ┆ A    ┆ 2021-01-01 │
        └────────────┴──────┴────────────┘

        A single time column mixing several formats is handled by coalescing lenient
        (``::?``) parses — each row takes the first format that matches, and rows
        matching none (garbage, ``""``, null) get a null time and are dropped (and
        counted in the WARNING). This is the documented multi-format idiom; a strict
        (``::"fmt"``) cast would instead *error* on the first unparsable value:

        >>> raw = pl.DataFrame({
        ...     "subject_id": [1, 2, 3, 4],
        ...     "ts": ["01/02/21 12:30:00", "01/03/21", "not a date", None],
        ... })
        >>> ev = EventConfig.parse("visit", {
        ...     "code": "VISIT",
        ...     "time": 'coalesce($ts::?"%m/%d/%y %H:%M:%S", $ts::?"%m/%d/%y")',
        ... })
        >>> ev.extract(raw.lazy(), "visits/visit").collect().select("subject_id", "time")
        shape: (2, 2)
        ┌────────────┬─────────────────────┐
        │ subject_id ┆ time                │
        │ ---        ┆ ---                 │
        │ i64        ┆ datetime[μs]        │
        ╞════════════╪═════════════════════╡
        │ 1          ┆ 2021-01-02 12:30:00 │
        │ 2          ┆ 2021-01-03 00:00:00 │
        └────────────┴─────────────────────┘

        A *composite / interpolated* code null-propagates the same way: if **any** referenced
        component is null, the whole code is null, so the row is dropped. Below, only the row
        with both ``$itemid`` and ``$valueuom`` present survives:

        >>> raw = pl.DataFrame({
        ...     "subject_id": [1, 2, 3, 4],
        ...     "itemid": ["GLU", "GLU", None, None],  # present, present, null, null
        ...     "valueuom": ["mg/dL", None, "mg/dL", None],  # present, null, present, null
        ... })
        >>> ev = EventConfig.parse("lab", {"code": 'f"{$itemid}//{$valueuom}"', "time": None})
        >>> ev.extract(raw.lazy(), "labs/lab").collect().select("subject_id", "code")
        shape: (1, 2)
        ┌────────────┬────────────┐
        │ subject_id ┆ code       │
        │ ---        ┆ ---        │
        │ i64        ┆ str        │
        ╞════════════╪════════════╡
        │ 1          ┆ GLU//mg/dL │
        └────────────┴────────────┘

        To keep rows with missing components, coalesce them in the MESSY expression with the
        dftly ``??`` operator (single-quote the literal fallback inside the f-string) — each
        null component is filled with your chosen literal, so all four rows are retained:

        >>> ev = EventConfig.parse(
        ...     "lab",
        ...     {"code": '''f"{$itemid ?? 'UNK'}//{$valueuom ?? 'UNK'}"''', "time": None},
        ... )
        >>> ev.extract(raw.lazy(), "labs/lab").collect().sort("subject_id").select(
        ...     "subject_id", "code"
        ... )
        shape: (4, 2)
        ┌────────────┬────────────┐
        │ subject_id ┆ code       │
        │ ---        ┆ ---        │
        │ i64        ┆ str        │
        ╞════════════╪════════════╡
        │ 1          ┆ GLU//mg/dL │
        │ 2          ┆ GLU//UNK   │
        │ 3          ┆ UNK//mg/dL │
        │ 4          ┆ UNK//UNK   │
        └────────────┴────────────┘

        The fallback is per-component and any literal you choose — coalescing only some
        components, or with distinct markers, are both fine (see the README's *Code
        Construction* section for more variations).

        With ``do_dedup_text_and_numeric=True``, a ``text_value`` that
        numerically equals ``numeric_value`` is nulled out:

        >>> raw = pl.DataFrame({
        ...     "subject_id": [1, 2],
        ...     "ts": ["2021-01-01", "2021-01-02"],
        ...     "val": [1.5, 2.0],
        ...     "text": ["1.5", "other"],
        ... })
        >>> ev = EventConfig.parse("m", {
        ...     "code": "MEAS",
        ...     "time": '$ts::"%Y-%m-%d"',
        ...     "numeric_value": "$val",
        ...     "text_value": "$text",
        ... })
        >>> ev.extract(raw.lazy(), "t/m", do_dedup_text_and_numeric=True).collect().select(
        ...     "numeric_value", "text_value"
        ... )
        shape: (2, 2)
        ┌───────────────┬────────────┐
        │ numeric_value ┆ text_value │
        │ ---           ┆ ---        │
        │ f64           ┆ str        │
        ╞═══════════════╪════════════╡
        │ 1.5           ┆ null       │
        │ 2.0           ┆ other      │
        └───────────────┴────────────┘

        The ``code_components`` struct's fields are named for the *source* columns the
        code references — including when a source column is literally named ``code``
        (the idiomatic ICD/OMOP vocabulary shape; ``extract_code_metadata`` unnests this
        struct and must alias the assembled code away from it):

        >>> raw = pl.DataFrame({"subject_id": [1], "code": ["250.00"], "ts": ["2020-01-01"]})
        >>> ev = EventConfig.parse("dx", {"code": 'f"ICD//{$code}"', "time": '$ts::"%Y-%m-%d"'})
        >>> ev.extract(raw.lazy(), "diagnoses/dx").collect().schema["code_components"]
        Struct({'code': String})

        A non-static event whose ``time`` expression is not temporally typed is rejected
        here, per event block — downstream schema alignment would otherwise silently
        reinterpret integers as 1970-epoch offsets, and strings would sort
        lexicographically (#207). Both the raw-integer-offset and the forgotten-cast
        string shapes are caught:

        >>> raw = pl.DataFrame({"subject_id": [1], "hr": [80.0], "offset": [1444]})
        >>> ev = EventConfig.parse("hr", {"code": "HR", "time": "$offset", "numeric_value": "$hr"})
        >>> ev.extract(raw.lazy(), "vitals/hr")
        Traceback (most recent call last):
            ...
        ValueError: `vitals/hr`: the `time` expression produced dtype Int64, not a date/datetime. ...
        >>> raw = pl.DataFrame({"subject_id": [1], "code": ["250.00"], "ts": ["2020-01-01"]})
        >>> ev = EventConfig.parse("dx", {"code": 'f"ICD//{$code}"', "time": "$ts"})
        >>> ev.extract(raw.lazy(), "diagnoses/dx")
        Traceback (most recent call last):
            ...
        ValueError: `diagnoses/dx`: the `time` expression produced dtype String, not a date/datetime. ...
    """
    exprs: dict[str, pl.Expr] = {"subject_id": pl.col("subject_id")}

    # `code` is the native dftly expression. String interpolation null-propagates: if any
    # referenced component is null the whole code is null, and the row is dropped below (a MEDS
    # code may not be null). Authors opt into retaining such rows by coalescing components in
    # the MESSY expression, e.g. `f"{$itemid ?? 'UNK'}//{$valueuom ?? 'UNK'}"`.
    exprs["code"] = self.polars_exprs["code"]
    if self.code_source_columns:
        # Raw, typed component values (the `code` string above is derived from these).
        exprs["code_components"] = pl.struct(
            **{col: pl.col(col) for col in sorted(self.code_source_columns)}
        )
    if self.self_metadata_exprs:
        # Evaluated ``_self`` metadata outputs, paired row-wise with the components
        # above. ``extract_code_metadata`` reads distinct component→metadata pairs
        # from these two structs; ``merge_to_MEDS_cohort`` drops both.
        exprs[METADATA_COMPONENTS_COL] = pl.struct(
            **{out: node.polars_expr for out, node in self.self_metadata_exprs.items()}
        )

    exprs["time"] = self.polars_exprs["time"]

    for k in self.columns:
        if k in ("code", "time"):
            continue
        exprs[k] = self.polars_exprs[k]

    if do_dedup_text_and_numeric and "numeric_value" in exprs and "text_value" in exprs:
        text_expr = exprs["text_value"]
        num_expr = exprs["numeric_value"]
        exprs["text_value"] = (
            pl.when(text_expr.cast(pl.Float32, strict=False) == num_expr.cast(pl.Float32))
            .then(pl.lit(None, pl.String))
            .otherwise(text_expr)
        )

    exprs[SOURCE_BLOCK_COL] = pl.lit(source_block)

    out = df.select(**exprs)

    # A non-static `time` must be temporally typed HERE, per event block, because nothing
    # downstream can catch the mistake intelligibly: DataSchema.align at finalize
    # reinterprets Int64 as 1970-epoch microseconds with no error (#207), merge's
    # diagonal_relaxed concat degrades sibling shards' correct Datetime columns to match a
    # wrong one, and a String time sorts lexicographically. Null passes: an all-null source
    # column resolves to dtype Null, and those rows are dropped (and counted) just below.
    if not self.is_static:
        time_dtype = out.collect_schema()["time"]
        if time_dtype != pl.Null and not isinstance(time_dtype, pl.Datetime | pl.Date):
            raise ValueError(
                f"`{source_block}`: the `time` expression produced dtype {time_dtype}, not a "
                "date/datetime. Integer offset columns must be converted to timestamps (e.g. a "
                "`_table.cols` pseudotime chain adding the offset to an anchor time), and string "
                'columns need an explicit cast (`$col::"%Y-%m-%d %H:%M:%S"`-style).'
            )

    # Null-`code` / null-`time` rows are dropped by filtering the COMPUTED columns, after the
    # select — never via a fallible predicate over the raw source expressions. Polars pushes
    # eligible predicates into the parquet scan itself, where they are evaluated on
    # validity-unmasked data (null slots of a String column surface as ""), so a predicate
    # containing a strict cast/strptime panics on polars >= 1.28 (and mis-evaluates silently
    # before that): https://github.com/pola-rs/polars/issues/28521. A predicate on a computed
    # column instead stays above the SELECT boundary (verified on polars 1.26-1.43 and
    # regression-guarded in tests), and the scan's projection pushdown is preserved. Note the
    # deliberate consequence for strict (`::"fmt"`) time casts: an unparsable non-null value
    # (e.g. "") now raises an InvalidOperationError naming the value, instead of being
    # silently pre-filtered; authors who want unparsable values dropped opt in with a
    # lenient (`::?"fmt"`) cast, which nulls them so they are dropped (and counted) here.
    drop_null_code = bool(self.code_source_columns)
    drop_null_time = not self.is_static

    if drop_null_code or drop_null_time:
        # Drop accounting: one aggregation pass over the pre-filter frame so silently
        # vanishing rows (unparsable times, null code components) are surfaced per event.
        stats_exprs = [pl.len().alias("n_total")]
        if drop_null_code:
            stats_exprs.append(pl.col("code").is_null().sum().alias("n_null_code"))
        if drop_null_time:
            stats_exprs.append(pl.col("time").is_null().sum().alias("n_null_time"))
        stats = out.select(stats_exprs).collect()
        n_total = stats["n_total"][0]
        n_null_code = stats["n_null_code"][0] if drop_null_code else 0
        n_null_time = stats["n_null_time"][0] if drop_null_time else 0
        if n_null_time or n_null_code:
            parts = []
            if n_null_time:
                parts.append(
                    f"{n_null_time}/{n_total} rows with null time "
                    f"(unparsable or missing under the configured formats)"
                )
            if n_null_code:
                count = f"{n_null_code}" if n_null_time else f"{n_null_code}/{n_total} rows"
                parts.append(f"{count} with null code")
            logger.warning(f"`{source_block}`: dropped " + " and ".join(parts))

    if drop_null_code:
        # A MEDS `code` may never be null. A bare-column code that is null, or an interpolated
        # code with a null (un-coalesced) component, evaluates to a null code — drop those
        # rows. Rows the author rescued with a `??` coalesce never evaluate to null, so they
        # are kept.
        out = out.filter(pl.col("code").is_not_null())
    if drop_null_time:
        out = out.filter(pl.col("time").is_not_null())

    return out.unique(maintain_order=True)

parse(name, raw) classmethod

Parse a raw event block into an EventConfig.

Each column value is compiled through :class:dftly.Parser, so raw input may be either a dftly expression string ("$col", 'f"PREFIX//{$col}"', "hash($col)") or an expanded dftly dict form. The time column is the only key that may be None — a None time produces a static event.

Examples:

Strings get parsed to nodes:

>>> ev = EventConfig.parse("lab", {"code": "X", "time": None, "numeric_value": "$v"})
>>> type(ev.columns["code"]).__name__
'Literal'
>>> type(ev.columns["numeric_value"]).__name__
'Column'
>>> ev.columns["time"] is None
True

Validation errors surface at parse time with the event name:

>>> EventConfig.parse("bad", {"time": None})
Traceback (most recent call last):
    ...
KeyError: "Event 'bad' must contain a 'code' key. Got: [time]."
>>> EventConfig.parse("bad", {"code": "X", "subject_id": "$sid"})
Traceback (most recent call last):
    ...
ValueError: Event 'bad' contains a 'subject_id' key. subject_id is a table-level concept ...

An underscore-prefixed key other than _metadata is almost certainly a typo of that reserved name, and is rejected up front rather than falling through to column parsing with a misleading error:

>>> EventConfig.parse("dob", {
...     "code": "BIRTH",
...     "time": "$dob",
...     "_metdata": {"d_items": {"itemid": "$itemid", "description": "$label"}},
... })
Traceback (most recent call last):
    ...
ValueError: Event 'dob' has unknown reserved key(s) ['_metdata']. The only reserved
key at the event level is '_metadata'. Output column names may not begin with an
underscore.

_metadata blocks are compiled and validated here too (via :func:compile_metadata_block, which documents the full error catalog), so every config mistake fires at parse time with the event and prefix named. A block on a literal code:

>>> EventConfig.parse("admit", {
...     "code": "ADMISSION",
...     "time": None,
...     "_metadata": {"adm_meta": {"description": "$title"}},
... })
Traceback (most recent call last):
    ...
ValueError: The code expression 'ADMISSION' is a literal: ... no components to match
metadata on. ...

A block that produces no join key, naming the components the event offers:

>>> EventConfig.parse("med", {
...     "code": 'f"{$medication_name}//{$dose}"',
...     "time": None,
...     "_metadata": {"med_classes": {"description": "$drug_class"}},
... })
Traceback (most recent call last):
    ...
ValueError: _metadata block (event 'med', metadata prefix 'med_classes') produces no
join-key columns: ... Component columns available on this event: ['dose',
'medication_name'] ...

A reserved output name:

>>> EventConfig.parse("chart", {
...     "code": 'f"CHART//{$itemid}"',
...     "time": None,
...     "_metadata": {"d_items": {"itemid": "$itemid", "code_template": "$label"}},
... })
Traceback (most recent call last):
    ...
ValueError: _metadata output column name(s) ['code_template'] are reserved: ...

A value that is not a dftly expression, named with its event and prefix:

>>> EventConfig.parse("lab", {
...     "code": "$test_name",
...     "time": None,
...     "_metadata": {
...         "lab_meta": {"test_name": "$test_name", "description": ["special_title", "title"]}
...     },
... })
Traceback (most recent call last):
    ...
ValueError: _metadata column 'description' (event 'lab', metadata prefix 'lab_meta')
failed to parse as a dftly expression: ...

And the one nuance on reserved names: a key named code is legal exactly when the code expression references a source column literally named code (the ICD/OMOP vocabulary shape) — it is a join key, never an output:

>>> ev = EventConfig.parse("dx", {
...     "code": 'f"ICD//{$code}"',
...     "time": None,
...     "_metadata": {"icd_meta": {"code": "$code", "description": "$long_title"}},
... })
>>> ev.metadata
{'icd_meta': {'code': '$code', 'description': '$long_title'}}

Source code in MEDS_extract/config.py

@classmethod
def parse(cls, name: str, raw: Mapping[str, Any]) -> EventConfig:
    """Parse a raw event block into an EventConfig.

    Each column value is compiled through :class:`dftly.Parser`, so raw
    input may be either a dftly expression string (``"$col"``,
    ``'f"PREFIX//{$col}"'``, ``"hash($col)"``) or an expanded dftly dict
    form. The time column is the only key that may be ``None`` — a
    ``None`` time produces a static event.

    Examples:
        Strings get parsed to nodes:

        >>> ev = EventConfig.parse("lab", {"code": "X", "time": None, "numeric_value": "$v"})
        >>> type(ev.columns["code"]).__name__
        'Literal'
        >>> type(ev.columns["numeric_value"]).__name__
        'Column'
        >>> ev.columns["time"] is None
        True

        Validation errors surface at parse time with the event name:

        >>> EventConfig.parse("bad", {"time": None})
        Traceback (most recent call last):
            ...
        KeyError: "Event 'bad' must contain a 'code' key. Got: [time]."
        >>> EventConfig.parse("bad", {"code": "X", "subject_id": "$sid"})
        Traceback (most recent call last):
            ...
        ValueError: Event 'bad' contains a 'subject_id' key. subject_id is a table-level concept ...

        An underscore-prefixed key other than ``_metadata`` is almost certainly a
        typo of that reserved name, and is rejected up front rather than falling
        through to column parsing with a misleading error:

        >>> EventConfig.parse("dob", {
        ...     "code": "BIRTH",
        ...     "time": "$dob",
        ...     "_metdata": {"d_items": {"itemid": "$itemid", "description": "$label"}},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: Event 'dob' has unknown reserved key(s) ['_metdata']. The only reserved
        key at the event level is '_metadata'. Output column names may not begin with an
        underscore.

        ``_metadata`` blocks are compiled and validated here too (via
        :func:`compile_metadata_block`, which documents the full error catalog), so
        every config mistake fires at parse time with the event and prefix named.
        A block on a literal code:

        >>> EventConfig.parse("admit", {
        ...     "code": "ADMISSION",
        ...     "time": None,
        ...     "_metadata": {"adm_meta": {"description": "$title"}},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: The code expression 'ADMISSION' is a literal: ... no components to match
        metadata on. ...

        A block that produces no join key, naming the components the event offers:

        >>> EventConfig.parse("med", {
        ...     "code": 'f"{$medication_name}//{$dose}"',
        ...     "time": None,
        ...     "_metadata": {"med_classes": {"description": "$drug_class"}},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: _metadata block (event 'med', metadata prefix 'med_classes') produces no
        join-key columns: ... Component columns available on this event: ['dose',
        'medication_name'] ...

        A reserved output name:

        >>> EventConfig.parse("chart", {
        ...     "code": 'f"CHART//{$itemid}"',
        ...     "time": None,
        ...     "_metadata": {"d_items": {"itemid": "$itemid", "code_template": "$label"}},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: _metadata output column name(s) ['code_template'] are reserved: ...

        A value that is not a dftly expression, named with its event and prefix:

        >>> EventConfig.parse("lab", {
        ...     "code": "$test_name",
        ...     "time": None,
        ...     "_metadata": {
        ...         "lab_meta": {"test_name": "$test_name", "description": ["special_title", "title"]}
        ...     },
        ... })
        Traceback (most recent call last):
            ...
        ValueError: _metadata column 'description' (event 'lab', metadata prefix 'lab_meta')
        failed to parse as a dftly expression: ...

        And the one nuance on reserved names: a key named ``code`` is legal exactly
        when the code expression references a source column literally named ``code``
        (the ICD/OMOP vocabulary shape) — it is a join key, never an output:

        >>> ev = EventConfig.parse("dx", {
        ...     "code": 'f"ICD//{$code}"',
        ...     "time": None,
        ...     "_metadata": {"icd_meta": {"code": "$code", "description": "$long_title"}},
        ... })
        >>> ev.metadata
        {'icd_meta': {'code': '$code', 'description': '$long_title'}}
    """
    raw = dict(raw)
    metadata = dict(raw.pop("_metadata", {}))

    stray_reserved = sorted(k for k in raw if k.startswith("_"))
    if stray_reserved:
        raise ValueError(
            f"Event '{name}' has unknown reserved key(s) {stray_reserved}. The only reserved "
            f"key at the event level is '_metadata'. Output column names may not begin with "
            f"an underscore."
        )

    raw_code = raw["code"] if isinstance(raw.get("code"), str) else None

    columns: dict[str, NodeBase | None] = {}
    parser = Parser()
    for k, v in raw.items():
        if k == "time" and v is None:
            columns[k] = None
            continue
        try:
            columns[k] = v if isinstance(v, NodeBase) else parser(v)
        except Exception as e:
            raise ValueError(f"Event '{name}' column '{k}' failed to parse: {e}") from e

    return cls(name=name, columns=columns, metadata=metadata, raw_code=raw_code)

JoinConfig dataclass

Parsed left-join configuration for a single table.

A join must always pull in at least one column from the joined table (cols is required) — a join without cols would be a no-op. The MESSY syntax has three forms:

Short form, for the common case of a shared key column::

join: {stays: {key: stay_id, cols: [patient_id, dischtime]}}

Long form, when the key columns differ::

join: {admissions: {left_on: hadm_id, right_on: admission_id, cols: [dischtime]}}

Every key field also accepts a list of columns, joined as a composite key (left_on and right_on pair up column-for-column)::

join: {parameters: {key: [table, label], cols: [unit]}}

A composite key doubles as a right-side row filter: put a constant on the left via _table.cols and include it in the key, and only the matching right rows join. The canonical case is a multi-row-per-id table where exactly one row kind is wanted::

_table:
  cols:
    drug_type: "'MAIN'"        # literal; materialized before the join
  join:
    hosp/prescriptions:
      key: [pharmacy_id, drug_type]
      cols: [ndc]

Join-key derived columns must be computable from raw source columns alone (validated at parse) because they are applied before the join runs.

Aggregated form, when the right-hand side needs a group_by + reduction before the join. The right-side rows are grouped by right_on and each named column is reduced with the listed aggregation — e.g. earliest death-time per subject from the admissions table::

join:
  hosp/admissions:
    key: subject_id
    cols:
      deathtime: min

The aggregated form exists so that reductions like this one stay in the MESSY spec instead of a bespoke pre-MEDS Python step.

Aggregated-form semantics worth knowing:

• Every supported aggregation is order-independent (min, max, sum, mean, count), so aggregated values are deterministic regardless of the order the right side’s files are scanned in. first/last are deliberately unsupported — see _JOIN_AGGREGATIONS.
• cols is either all flat (list form) or all aggregated (mapping form); mixing is not expressible because the aggregated form groups the entire right side. To pull both flat and aggregated columns from one table you currently need the flat join plus a derived expression, or a pre-MEDS step.
• min/max on a String column compares lexicographically — right for ISO-8601-style timestamps, silently wrong for e.g. %m/%d/%Y; :meth:apply warns at runtime. sum/mean on a String column are rejected at runtime.
• Every aggregated join logs one WARNING at construction time: aggregation folds multiple source rows into one value, which can silently absorb data errors (conflicting values are resolved by the aggregation instead of surfacing) and makes row-level provenance untraceable through the join. The warning names the join and its col→agg pairs so the reduction is a deliberate, visible choice.
• If a code expression references an aggregated column, the value in code_components is the aggregate. For min/max that is still a real raw value from the right table, so _metadata component matching stays coherent; for sum/mean/count it is synthetic and will match nothing in a raw-valued metadata table.

Examples:

Plain-list cols — no aggregation. Fields are shown individually rather than via the default repr, which would otherwise push the example past the 110-col line limit:

>>> jc = JoinConfig.parse({"stays": {"key": "stay_id", "cols": ["subject_id"]}})
>>> jc.input_prefix, jc.left_on, jc.right_on, jc.cols, jc.aggregations
('stays', ('stay_id',), ('stay_id',), ('subject_id',), ())
>>> jc = JoinConfig.parse(
...     {"admissions": {"left_on": "hadm_id", "right_on": "adm_id", "cols": ["dischtime"]}}
... )
>>> jc.left_on, jc.right_on, jc.cols
(('hadm_id',), ('adm_id',), ('dischtime',))

Composite keys: key / left_on / right_on accept a list of columns, joined as a unit. Sides must pair up column-for-column:

>>> jc = JoinConfig.parse(
...     {"prescriptions": {"key": ["pharmacy_id", "drug_type"], "cols": ["ndc"]}}
... )
>>> jc.left_on, jc.right_on
(('pharmacy_id', 'drug_type'), ('pharmacy_id', 'drug_type'))
>>> JoinConfig.parse(
...     {"p": {"left_on": ["a", "b"], "right_on": "a", "cols": ["x"]}}
... )
Traceback (most recent call last):
    ...
ValueError: Join config for 'p': 'left_on' and 'right_on' must pair up column-for-column, ...
>>> JoinConfig.parse({"p": {"key": [], "cols": ["x"]}})
Traceback (most recent call last):
    ...
ValueError: Join config for 'p': 'left_on' must be a non-empty column name or a non-empty ...
>>> JoinConfig.parse({"p": {"key": ["a", "a"], "cols": ["x"]}})
Traceback (most recent call last):
    ...
ValueError: Join config for 'p': 'left_on' lists the same column more than once: ['a', 'a'].

Aggregated form — cols becomes a {name: agg} mapping. Order is preserved from the YAML document:

>>> jc = JoinConfig.parse({
...     "hosp/admissions": {
...         "key": "subject_id",
...         "cols": {"deathtime": "min", "admittime": "max"},
...     }
... })
>>> jc.cols
('deathtime', 'admittime')
>>> jc.aggregations
(('deathtime', 'min'), ('admittime', 'max'))

Validation catches the usual shapes:

>>> JoinConfig.parse({"a": {}, "b": {}})
Traceback (most recent call last):
    ...
ValueError: Join config must have exactly one key (the input prefix), got: ['a', 'b']
>>> JoinConfig.parse({"stays": {"cols": ["dischtime"]}})
Traceback (most recent call last):
    ...
ValueError: Join config for 'stays' must specify either 'key' or both 'left_on' and 'right_on'.
>>> JoinConfig.parse({"stays": {"key": "stay_id"}})
Traceback (most recent call last):
    ...
ValueError: Join config for 'stays' must pull in at least one column via 'cols'.

Unknown keys are rejected, not dropped — joins are always left joins, so polars muscle-memory like how: inner must fail loudly rather than silently produce a left join:

>>> JoinConfig.parse({"stays": {"key": "stay_id", "cols": ["dischtime"], "how": "inner"}})
Traceback (most recent call last):
    ...
ValueError: Join config for 'stays' has unknown keys: ['how']. Allowed keys: 'key',
'left_on', 'right_on', 'cols'. (Joins are always left joins — there is no 'how'.)

key may not be combined with left_on/right_on — the short and long forms are mutually exclusive, so a leftover key from editing cannot silently win over the explicit column pair:

>>> JoinConfig.parse(
...     {"stays": {"key": "stay_id", "left_on": "hadm_id", "right_on": "adm_id", "cols": ["x"]}}
... )
Traceback (most recent call last):
    ...
ValueError: Join config for 'stays' specifies both 'key' and 'left_on'/'right_on'. Use 'key'
alone when both sides share one column name, or 'left_on' + 'right_on' when the names differ.

Unknown aggregation names are rejected eagerly — typos beat silent wrong-results:

>>> JoinConfig.parse({
...     "stays": {"key": "subject_id", "cols": {"deathtime": "median"}}
... })
Traceback (most recent call last):
    ...
ValueError: Join config for 'stays' col 'deathtime': unsupported aggregation 'median'. Supported: ...

first/last are rejected too — they are scan-order-dependent over the unordered multi-file right side (use min/max over an ordering column instead):

>>> JoinConfig.parse({
...     "stays": {"key": "subject_id", "cols": {"deathtime": "first"}}
... })
Traceback (most recent call last):
    ...
ValueError: Join config for 'stays' col 'deathtime': unsupported aggregation 'first'. Supported: ...

Writing the aggregated form as a YAML list of one-entry mappings (- deathtime: min) is a likely slip and gets a targeted message:

>>> JoinConfig.parse({
...     "stays": {"key": "subject_id", "cols": [{"deathtime": "min"}]}
... })
Traceback (most recent call last):
    ...
ValueError: Join config for 'stays': 'cols' is a list containing mappings ... a single mapping ...

Validation holds at direct construction, not just through parse:

>>> JoinConfig(
...     input_prefix="stays", left_on="sid", right_on="sid",
...     cols=("deathtime",), aggregations=(("deathtime", "median"),),
... )
Traceback (most recent call last):
    ...
ValueError: Join config for 'stays' col 'deathtime': unsupported aggregation 'median'. Supported: ...

Source code in MEDS_extract/config.py

@dataclass(frozen=True)
class JoinConfig:
    """Parsed left-join configuration for a single table.

    A join must always pull in at least one column from the joined table
    (``cols`` is required) — a join without ``cols`` would be a no-op. The
    MESSY syntax has three forms:

    Short form, for the common case of a shared key column::

        join: {stays: {key: stay_id, cols: [patient_id, dischtime]}}

    Long form, when the key columns differ::

        join: {admissions: {left_on: hadm_id, right_on: admission_id, cols: [dischtime]}}

    Every key field also accepts a **list of columns**, joined as a composite key
    (``left_on`` and ``right_on`` pair up column-for-column)::

        join: {parameters: {key: [table, label], cols: [unit]}}

    A composite key doubles as a right-side row filter: put a constant on the left
    via ``_table.cols`` and include it in the key, and only the matching right rows
    join. The canonical case is a multi-row-per-id table where exactly one row kind
    is wanted::

        _table:
          cols:
            drug_type: "'MAIN'"        # literal; materialized before the join
          join:
            hosp/prescriptions:
              key: [pharmacy_id, drug_type]
              cols: [ndc]

    Join-key derived columns must be computable from raw source columns alone
    (validated at parse) because they are applied before the join runs.

    Aggregated form, when the right-hand side needs a ``group_by`` + reduction
    before the join. The right-side rows are grouped by ``right_on`` and each
    named column is reduced with the listed aggregation — e.g. earliest
    death-time per subject from the admissions table::

        join:
          hosp/admissions:
            key: subject_id
            cols:
              deathtime: min

    The aggregated form exists so that reductions like this one stay in the
    MESSY spec instead of a bespoke pre-MEDS Python step.

    Aggregated-form semantics worth knowing:

    - Every supported aggregation is **order-independent** (``min``, ``max``,
      ``sum``, ``mean``, ``count``), so aggregated values are deterministic
      regardless of the order the right side's files are scanned in.
      ``first``/``last`` are deliberately unsupported — see
      ``_JOIN_AGGREGATIONS``.
    - ``cols`` is either *all* flat (list form) or *all* aggregated (mapping
      form); mixing is not expressible because the aggregated form groups the
      entire right side. To pull both flat and aggregated columns from one
      table you currently need the flat join plus a derived expression, or a
      pre-MEDS step.
    - ``min``/``max`` on a String column compares **lexicographically** —
      right for ISO-8601-style timestamps, silently wrong for e.g.
      ``%m/%d/%Y``; :meth:`apply` warns at runtime. ``sum``/``mean`` on a
      String column are rejected at runtime.
    - Every aggregated join logs one WARNING at construction time: aggregation
      folds multiple source rows into one value, which can silently absorb data
      errors (conflicting values are resolved by the aggregation instead of
      surfacing) and makes row-level provenance untraceable through the join.
      The warning names the join and its col→agg pairs so the reduction is a
      deliberate, visible choice.
    - If a ``code`` expression references an aggregated column, the value in
      ``code_components`` is the *aggregate*. For ``min``/``max`` that is
      still a real raw value from the right table, so ``_metadata`` component
      matching stays coherent; for ``sum``/``mean``/``count`` it is synthetic
      and will match nothing in a raw-valued metadata table.

    Examples:
        Plain-list ``cols`` — no aggregation. Fields are shown individually
        rather than via the default repr, which would otherwise push the
        example past the 110-col line limit:

        >>> jc = JoinConfig.parse({"stays": {"key": "stay_id", "cols": ["subject_id"]}})
        >>> jc.input_prefix, jc.left_on, jc.right_on, jc.cols, jc.aggregations
        ('stays', ('stay_id',), ('stay_id',), ('subject_id',), ())
        >>> jc = JoinConfig.parse(
        ...     {"admissions": {"left_on": "hadm_id", "right_on": "adm_id", "cols": ["dischtime"]}}
        ... )
        >>> jc.left_on, jc.right_on, jc.cols
        (('hadm_id',), ('adm_id',), ('dischtime',))

        Composite keys: ``key`` / ``left_on`` / ``right_on`` accept a list of columns,
        joined as a unit. Sides must pair up column-for-column:

        >>> jc = JoinConfig.parse(
        ...     {"prescriptions": {"key": ["pharmacy_id", "drug_type"], "cols": ["ndc"]}}
        ... )
        >>> jc.left_on, jc.right_on
        (('pharmacy_id', 'drug_type'), ('pharmacy_id', 'drug_type'))
        >>> JoinConfig.parse(
        ...     {"p": {"left_on": ["a", "b"], "right_on": "a", "cols": ["x"]}}
        ... )  # doctest: +ELLIPSIS
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'p': 'left_on' and 'right_on' must pair up column-for-column, ...
        >>> JoinConfig.parse({"p": {"key": [], "cols": ["x"]}})  # doctest: +ELLIPSIS
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'p': 'left_on' must be a non-empty column name or a non-empty ...
        >>> JoinConfig.parse({"p": {"key": ["a", "a"], "cols": ["x"]}})  # doctest: +ELLIPSIS
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'p': 'left_on' lists the same column more than once: ['a', 'a'].

        Aggregated form — ``cols`` becomes a ``{name: agg}`` mapping. Order is
        preserved from the YAML document:

        >>> jc = JoinConfig.parse({
        ...     "hosp/admissions": {
        ...         "key": "subject_id",
        ...         "cols": {"deathtime": "min", "admittime": "max"},
        ...     }
        ... })
        >>> jc.cols
        ('deathtime', 'admittime')
        >>> jc.aggregations
        (('deathtime', 'min'), ('admittime', 'max'))

        Validation catches the usual shapes:

        >>> JoinConfig.parse({"a": {}, "b": {}})
        Traceback (most recent call last):
            ...
        ValueError: Join config must have exactly one key (the input prefix), got: ['a', 'b']
        >>> JoinConfig.parse({"stays": {"cols": ["dischtime"]}})
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'stays' must specify either 'key' or both 'left_on' and 'right_on'.
        >>> JoinConfig.parse({"stays": {"key": "stay_id"}})
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'stays' must pull in at least one column via 'cols'.

        Unknown keys are rejected, not dropped — joins are always left joins, so
        polars muscle-memory like ``how: inner`` must fail loudly rather than
        silently produce a left join:

        >>> JoinConfig.parse({"stays": {"key": "stay_id", "cols": ["dischtime"], "how": "inner"}})
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'stays' has unknown keys: ['how']. Allowed keys: 'key',
        'left_on', 'right_on', 'cols'. (Joins are always left joins — there is no 'how'.)

        ``key`` may not be combined with ``left_on``/``right_on`` — the short and
        long forms are mutually exclusive, so a leftover key from editing cannot
        silently win over the explicit column pair:

        >>> JoinConfig.parse(
        ...     {"stays": {"key": "stay_id", "left_on": "hadm_id", "right_on": "adm_id", "cols": ["x"]}}
        ... )
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'stays' specifies both 'key' and 'left_on'/'right_on'. Use 'key'
        alone when both sides share one column name, or 'left_on' + 'right_on' when the names differ.

        Unknown aggregation names are rejected eagerly — typos beat silent
        wrong-results:

        >>> JoinConfig.parse({  # doctest: +ELLIPSIS
        ...     "stays": {"key": "subject_id", "cols": {"deathtime": "median"}}
        ... })
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'stays' col 'deathtime': unsupported aggregation 'median'. Supported: ...

        ``first``/``last`` are rejected too — they are scan-order-dependent
        over the unordered multi-file right side (use ``min``/``max`` over an
        ordering column instead):

        >>> JoinConfig.parse({  # doctest: +ELLIPSIS
        ...     "stays": {"key": "subject_id", "cols": {"deathtime": "first"}}
        ... })
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'stays' col 'deathtime': unsupported aggregation 'first'. Supported: ...

        Writing the aggregated form as a YAML *list* of one-entry mappings
        (``- deathtime: min``) is a likely slip and gets a targeted message:

        >>> JoinConfig.parse({  # doctest: +ELLIPSIS
        ...     "stays": {"key": "subject_id", "cols": [{"deathtime": "min"}]}
        ... })
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'stays': 'cols' is a list containing mappings ... a single mapping ...

        Validation holds at direct construction, not just through ``parse``:

        >>> JoinConfig(  # doctest: +ELLIPSIS
        ...     input_prefix="stays", left_on="sid", right_on="sid",
        ...     cols=("deathtime",), aggregations=(("deathtime", "median"),),
        ... )
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'stays' col 'deathtime': unsupported aggregation 'median'. Supported: ...
    """

    input_prefix: str
    left_on: tuple[str, ...]
    right_on: tuple[str, ...]
    cols: tuple[str, ...]
    aggregations: tuple[tuple[str, str], ...] = ()

    def __post_init__(self):
        # Key normalization at construction: a scalar means a single-column key, a
        # sequence a composite key. Everything downstream (the join itself, the
        # aggregation group_by, and the column planner) iterates the tuples.
        for side in ("left_on", "right_on"):
            v = getattr(self, side)
            object.__setattr__(self, side, (v,) if isinstance(v, str) else tuple(v))
        for side in ("left_on", "right_on"):
            keys = getattr(self, side)
            if not keys or any(not isinstance(k, str) or not k for k in keys):
                raise ValueError(
                    f"Join config for '{self.input_prefix}': {side!r} must be a non-empty column "
                    f"name or a non-empty list of them, got {keys!r}."
                )
            if len(set(keys)) != len(keys):
                raise ValueError(
                    f"Join config for '{self.input_prefix}': {side!r} lists the same column more "
                    f"than once: {list(keys)!r}."
                )
        if len(self.left_on) != len(self.right_on):
            raise ValueError(
                f"Join config for '{self.input_prefix}': 'left_on' and 'right_on' must pair up "
                f"column-for-column, got {len(self.left_on)} left key(s) {list(self.left_on)!r} vs "
                f"{len(self.right_on)} right key(s) {list(self.right_on)!r}."
            )
        if not self.cols:
            raise ValueError(
                f"Join config for '{self.input_prefix}' must pull in at least one column via 'cols'."
            )
        # A right key whose left counterpart has a different name is coalesced into
        # that counterpart by the left join and delivers no column of its own, so
        # listing it in ``cols`` promises a column that can never arrive.
        renamed_keys = {r: lft for lft, r in zip(self.left_on, self.right_on, strict=False) if lft != r}
        dropped = sorted(c for c in self.cols if c in renamed_keys)
        if dropped:
            hints = ", ".join(f"'{c}' arrives as '{renamed_keys[c]}'" for c in dropped)
            raise ValueError(
                f"Join config for '{self.input_prefix}': 'cols' lists right key column(s) "
                f"{dropped}, but a left join coalesces each right key into its left counterpart "
                f"({hints}) and delivers no column under the right key's name. Reference the "
                f"left key column instead — the values are identical after the join."
            )
        # Aggregation validation lives here (not only in ``parse``) so a directly-constructed
        # JoinConfig is held to the same rules — validation-at-construction, per repo convention.
        for col, agg in self.aggregations:
            if not isinstance(agg, str) or agg not in _JOIN_AGGREGATIONS:
                supported = ", ".join(sorted(_JOIN_AGGREGATIONS))
                raise ValueError(
                    f"Join config for '{self.input_prefix}' col {col!r}: "
                    f"unsupported aggregation {agg!r}. Supported: {supported}."
                )
        if self.aggregations and tuple(col for col, _ in self.aggregations) != self.cols:
            raise ValueError(
                f"Join config for '{self.input_prefix}': 'aggregations' must cover exactly the "
                f"columns in 'cols', in order — got cols={self.cols!r} but aggregations over "
                f"{tuple(col for col, _ in self.aggregations)!r}. (Mixing flat and aggregated "
                f"columns in one join is not supported: the aggregated form groups the whole "
                f"right side.)"
            )
        if self.aggregations:
            agg_desc = ", ".join(f"{agg}({col})" for col, agg in self.aggregations)
            logger.warning(
                f"Join config for '{self.input_prefix}': aggregated join ({agg_desc}) — "
                f"aggregations fold multiple source rows into one value, so data conflicts "
                f"are resolved silently (e.g. by min/max instead of surfacing) and row-level "
                f"provenance is not traceable through the aggregation. Use with care."
            )

    @classmethod
    def parse(cls, raw: Mapping[str, Any]) -> JoinConfig:
        if not isinstance(raw, dict) or len(raw) != 1:
            got = sorted(raw.keys()) if isinstance(raw, dict) else raw
            raise ValueError(f"Join config must have exactly one key (the input prefix), got: {got}")

        input_prefix, inner = next(iter(raw.items()))
        if not isinstance(inner, dict):
            raise ValueError(
                f"Join config for '{input_prefix}' must be a mapping with 'key'/'left_on'+'right_on' "
                f"and 'cols', got {type(inner).__name__}."
            )

        unknown_keys = set(inner) - {"key", "left_on", "right_on", "cols"}
        if unknown_keys:
            raise ValueError(
                f"Join config for '{input_prefix}' has unknown keys: {sorted(unknown_keys)}. "
                f"Allowed keys: 'key', 'left_on', 'right_on', 'cols'. (Joins are always left "
                f"joins — there is no 'how'.)"
            )
        if "key" in inner and ("left_on" in inner or "right_on" in inner):
            raise ValueError(
                f"Join config for '{input_prefix}' specifies both 'key' and 'left_on'/'right_on'. "
                f"Use 'key' alone when both sides share one column name, or 'left_on' + 'right_on' "
                f"when the names differ."
            )

        if "key" in inner:
            left_on = right_on = inner["key"]
        elif "left_on" in inner and "right_on" in inner:
            left_on = inner["left_on"]
            right_on = inner["right_on"]
        else:
            raise ValueError(
                f"Join config for '{input_prefix}' must specify either 'key' or both "
                f"'left_on' and 'right_on'."
            )

        cols_raw = inner.get("cols", ())
        cols, aggregations = cls._parse_cols(input_prefix, cols_raw)

        return cls(
            input_prefix=input_prefix,
            left_on=left_on,
            right_on=right_on,
            cols=cols,
            aggregations=aggregations,
        )

    @staticmethod
    def _parse_cols(input_prefix: str, cols_raw: Any) -> tuple[tuple[str, ...], tuple[tuple[str, str], ...]]:
        """Normalize ``cols`` in either plain-list or aggregation-mapping form.

        Returns ``(cols, aggregations)`` — ``aggregations`` is empty for the
        list form, populated from the mapping form. Kept separate from
        :meth:`parse` so the type-dispatch logic is easy to read.
        """
        # Mapping form: ``cols: {deathtime: min, admittime: max}``. Aggregation-name
        # validation happens in ``__post_init__`` (validation-at-construction), so unknown
        # names still surface at parse time, not at the first ``group_by().agg()`` call
        # downstream.
        if isinstance(cols_raw, dict):
            for col in cols_raw:
                if not isinstance(col, str):
                    raise ValueError(
                        f"Join config for '{input_prefix}' aggregation column names must be strings, "
                        f"got {type(col).__name__}: {col!r}."
                    )
            return tuple(cols_raw), tuple(cols_raw.items())

        # A likely YAML slip: writing the aggregated form as a *list* of one-entry
        # mappings (``cols:`` / ``- deathtime: min``) instead of one mapping. YAML parses
        # that as ``[{"deathtime": "min"}]`` — catch it with a targeted message before the
        # generic list-form error below garbles the intent.
        if isinstance(cols_raw, list | tuple) and any(isinstance(c, dict) for c in cols_raw):
            raise ValueError(
                f"Join config for '{input_prefix}': 'cols' is a list containing mappings "
                f"({cols_raw!r}). For an aggregated join, write 'cols' as a single mapping — "
                f"'cols: {{deathtime: min}}' or an indented block WITHOUT '-' list markers — "
                f"not a YAML list of '- name: agg' items."
            )

        # Plain-list form: ``cols: [patient_id, dischtime]``. No aggregation.
        if not isinstance(cols_raw, list | tuple) or any(not isinstance(c, str) for c in cols_raw):
            raise ValueError(
                f"Join config for '{input_prefix}' must specify 'cols' as a list of column-name "
                f"strings (or a {{name: aggregation}} mapping for grouped joins), got "
                f"{type(cols_raw).__name__}: {cols_raw!r}. A bare string like "
                f"'cols: subject_id' would silently be treated as a tuple of characters — "
                f"use 'cols: [subject_id]' instead."
            )
        return tuple(cols_raw), ()

    def apply(self, left: pl.LazyFrame, input_dir: Path | UPath) -> pl.LazyFrame:
        """Scan join-target files under ``input_dir`` and left-join them to ``left``.

        File resolution goes through :func:`MEDS_extract.io.resolve_source_files`,
        so every stage that applies a join uses the same layout-detection logic
        as the stages that read the main table. When ``aggregations`` is
        non-empty, the right-hand side is grouped by ``right_on`` and each
        named column is reduced before the join.

        Every joined column must arrive under a name the left table does not
        already have: polars would deliver a colliding column under a
        ``_right``-suffixed name, which MESSY expressions and the source-column
        plan cannot model, so the collision is rejected here with the offending
        names instead. (Same-named key columns are exempt — a left join
        coalesces them into one column.)

        Examples:
            End-to-end aggregated join — the admissions side has three rows for
            subject 1 (three admissions) and two rows for subject 2; the join
            pulls in the minimum ``deathtime`` per subject, fanning out over the
            patients table only once per subject:

            >>> _ = pl.Config.set_tbl_width_chars(600)
            >>> with yaml_disk('''
            ... hosp/admissions.parquet:
            ...   subject_id: [1, 1, 1, 2, 2]
            ...   deathtime:  ["2020-03-01", "2020-03-05", null, null, null]
            ... ''') as d:
            ...     jc = JoinConfig.parse({
            ...         "hosp/admissions": {"key": "subject_id", "cols": {"deathtime": "min"}}
            ...     })
            ...     patients = pl.LazyFrame({"subject_id": [1, 2, 3]})
            ...     jc.apply(patients, Path(d)).sort("subject_id").collect()
            shape: (3, 2)
            ┌────────────┬────────────┐
            │ subject_id ┆ deathtime  │
            │ ---        ┆ ---        │
            │ i64        ┆ str        │
            ╞════════════╪════════════╡
            │ 1          ┆ 2020-03-01 │
            │ 2          ┆ null       │
            │ 3          ┆ null       │
            └────────────┴────────────┘

            A joined column colliding with a left column is rejected:

            >>> with yaml_disk('''
            ... stays.parquet: {stay_id: [1], age: [40]}
            ... ''') as d:
            ...     jc = JoinConfig.parse({"stays": {"key": "stay_id", "cols": ["age"]}})
            ...     jc.apply(pl.LazyFrame({"stay_id": [1], "age": [39]}), Path(d))
            Traceback (most recent call last):
                ...
            ValueError: Join config for 'stays': joined column(s) ['age'] already exist on the left
            table. Polars would deliver them under '_right'-suffixed names, which MESSY expressions
            and the source-column plan cannot model, so plans and references would silently
            mis-resolve. Joined columns must arrive under names the left table does not already
            have: rename the column in the source data, or compute the value in a pre-processing
            step.
        """
        left_schema = left.collect_schema().names()
        coalesced_keys = {r for lft, r in zip(self.left_on, self.right_on, strict=False) if lft == r}
        colliding = [c for c in self.cols if c in left_schema and c not in coalesced_keys]
        if colliding:
            raise ValueError(
                f"Join config for '{self.input_prefix}': joined column(s) {colliding} already "
                f"exist on the left table. Polars would deliver them under '_right'-suffixed "
                f"names, which MESSY expressions and the source-column plan cannot model, so "
                f"plans and references would silently mis-resolve. Joined columns must arrive "
                f"under names the left table does not already have: rename the column in the "
                f"source data, or compute the value in a pre-processing step."
            )
        right = scan_source(resolve_source_files(input_dir, self.input_prefix))
        if self.aggregations:
            right = self._aggregate(right)
        # ``maintain_order`` pins the join output to left-row order (ties broken by right
        # order). The in-memory engine happens to preserve left order anyway (this is a
        # no-op there), but the streaming engine — which convert_to_subject_sharded's
        # sink-based write runs on — reorders nondeterministically without it, and merge's
        # stable sort propagates that order into the final MEDS bytes for same-time events.
        return left.join(
            right,
            left_on=list(self.left_on),
            right_on=list(self.right_on),
            how="left",
            maintain_order="left_right",
        )

    def _aggregate(self, right: pl.LazyFrame) -> pl.LazyFrame:
        """Group the right side by ``right_on`` and reduce each aggregated column.

        Column presence and String-dtype hazards are checked against the resolved scan
        schema *here*, so failures name the join table and column instead of surfacing
        as a bare polars error deep inside a stage. Two String-dtype cases get special
        treatment (live today because csv schema inference leaves datetime-like strings
        as String):

        - ``sum``/``mean`` on a String column is rejected: polars' own behavior is a
          cryptic ``InvalidOperationError`` for ``sum`` and — worse — a *silent all-null
          result* for ``mean``.
        - ``min``/``max`` on a String column warns: comparison is lexicographic, which
          is correct for ISO-8601-style timestamps but silently wrong for formats like
          ``%m/%d/%Y``.

        Examples:
            >>> jc = JoinConfig.parse({"adm": {"key": "sid", "cols": {"deathtime": "min"}}})
            >>> jc._aggregate(pl.LazyFrame({"sid": [1], "other": [2]}))
            Traceback (most recent call last):
                ...
            ValueError: Join target 'adm' is missing column(s) ['deathtime'] needed by the aggregated
            join (group key + aggregation inputs). Available columns: ['other', 'sid'].
            >>> jc = JoinConfig.parse({"adm": {"key": "sid", "cols": {"cost": "mean"}}})
            >>> jc._aggregate(pl.LazyFrame({"sid": [1], "cost": ["12.5"]}))
            Traceback (most recent call last):
                ...
            ValueError: Join config for 'adm': aggregation 'mean' on String column 'cost' would
            silently produce all-null results. Cast the source data to a numeric type (or use
            min/max/count).
        """
        schema = right.collect_schema()
        needed = [*self.right_on, *(col for col, _ in self.aggregations)]
        missing = [c for c in needed if c not in schema]
        if missing:
            raise ValueError(
                f"Join target '{self.input_prefix}' is missing column(s) {missing} needed by "
                f"the aggregated join (group key + aggregation inputs). Available columns: "
                f"{sorted(schema.names())}."
            )
        for col, agg in self.aggregations:
            if schema[col] == pl.String:
                if agg in ("sum", "mean"):
                    hazard = (
                        "would silently produce all-null results"
                        if agg == "mean"
                        else "is not supported by polars"
                    )
                    raise ValueError(
                        f"Join config for '{self.input_prefix}': aggregation {agg!r} on String "
                        f"column {col!r} {hazard}. Cast the source data to a numeric type "
                        f"(or use min/max/count)."
                    )
                if agg in ("min", "max"):
                    logger.warning(
                        f"Join config for '{self.input_prefix}': {agg!r} on String column "
                        f"{col!r} compares lexicographically. That is correct for "
                        f"ISO-8601-style timestamps ('%Y-%m-%d...') but silently wrong for "
                        f"formats like '%m/%d/%Y'. Verify this column's text ordering matches "
                        f"the intended ordering."
                    )
        agg_exprs = [getattr(pl.col(col), agg)() for col, agg in self.aggregations]
        return right.group_by(list(self.right_on)).agg(*agg_exprs)

_aggregate(right)

Group the right side by right_on and reduce each aggregated column.

Column presence and String-dtype hazards are checked against the resolved scan schema here, so failures name the join table and column instead of surfacing as a bare polars error deep inside a stage. Two String-dtype cases get special treatment (live today because csv schema inference leaves datetime-like strings as String):

• sum/mean on a String column is rejected: polars’ own behavior is a cryptic InvalidOperationError for sum and — worse — a silent all-null result for mean.
• min/max on a String column warns: comparison is lexicographic, which is correct for ISO-8601-style timestamps but silently wrong for formats like %m/%d/%Y.

Examples:

>>> jc = JoinConfig.parse({"adm": {"key": "sid", "cols": {"deathtime": "min"}}})
>>> jc._aggregate(pl.LazyFrame({"sid": [1], "other": [2]}))
Traceback (most recent call last):
    ...
ValueError: Join target 'adm' is missing column(s) ['deathtime'] needed by the aggregated
join (group key + aggregation inputs). Available columns: ['other', 'sid'].
>>> jc = JoinConfig.parse({"adm": {"key": "sid", "cols": {"cost": "mean"}}})
>>> jc._aggregate(pl.LazyFrame({"sid": [1], "cost": ["12.5"]}))
Traceback (most recent call last):
    ...
ValueError: Join config for 'adm': aggregation 'mean' on String column 'cost' would
silently produce all-null results. Cast the source data to a numeric type (or use
min/max/count).

Source code in MEDS_extract/config.py

def _aggregate(self, right: pl.LazyFrame) -> pl.LazyFrame:
    """Group the right side by ``right_on`` and reduce each aggregated column.

    Column presence and String-dtype hazards are checked against the resolved scan
    schema *here*, so failures name the join table and column instead of surfacing
    as a bare polars error deep inside a stage. Two String-dtype cases get special
    treatment (live today because csv schema inference leaves datetime-like strings
    as String):

    - ``sum``/``mean`` on a String column is rejected: polars' own behavior is a
      cryptic ``InvalidOperationError`` for ``sum`` and — worse — a *silent all-null
      result* for ``mean``.
    - ``min``/``max`` on a String column warns: comparison is lexicographic, which
      is correct for ISO-8601-style timestamps but silently wrong for formats like
      ``%m/%d/%Y``.

    Examples:
        >>> jc = JoinConfig.parse({"adm": {"key": "sid", "cols": {"deathtime": "min"}}})
        >>> jc._aggregate(pl.LazyFrame({"sid": [1], "other": [2]}))
        Traceback (most recent call last):
            ...
        ValueError: Join target 'adm' is missing column(s) ['deathtime'] needed by the aggregated
        join (group key + aggregation inputs). Available columns: ['other', 'sid'].
        >>> jc = JoinConfig.parse({"adm": {"key": "sid", "cols": {"cost": "mean"}}})
        >>> jc._aggregate(pl.LazyFrame({"sid": [1], "cost": ["12.5"]}))
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'adm': aggregation 'mean' on String column 'cost' would
        silently produce all-null results. Cast the source data to a numeric type (or use
        min/max/count).
    """
    schema = right.collect_schema()
    needed = [*self.right_on, *(col for col, _ in self.aggregations)]
    missing = [c for c in needed if c not in schema]
    if missing:
        raise ValueError(
            f"Join target '{self.input_prefix}' is missing column(s) {missing} needed by "
            f"the aggregated join (group key + aggregation inputs). Available columns: "
            f"{sorted(schema.names())}."
        )
    for col, agg in self.aggregations:
        if schema[col] == pl.String:
            if agg in ("sum", "mean"):
                hazard = (
                    "would silently produce all-null results"
                    if agg == "mean"
                    else "is not supported by polars"
                )
                raise ValueError(
                    f"Join config for '{self.input_prefix}': aggregation {agg!r} on String "
                    f"column {col!r} {hazard}. Cast the source data to a numeric type "
                    f"(or use min/max/count)."
                )
            if agg in ("min", "max"):
                logger.warning(
                    f"Join config for '{self.input_prefix}': {agg!r} on String column "
                    f"{col!r} compares lexicographically. That is correct for "
                    f"ISO-8601-style timestamps ('%Y-%m-%d...') but silently wrong for "
                    f"formats like '%m/%d/%Y'. Verify this column's text ordering matches "
                    f"the intended ordering."
                )
    agg_exprs = [getattr(pl.col(col), agg)() for col, agg in self.aggregations]
    return right.group_by(list(self.right_on)).agg(*agg_exprs)

_parse_cols(input_prefix, cols_raw) staticmethod

Normalize cols in either plain-list or aggregation-mapping form.

Returns (cols, aggregations) — aggregations is empty for the list form, populated from the mapping form. Kept separate from :meth:parse so the type-dispatch logic is easy to read.

Source code in MEDS_extract/config.py

@staticmethod
def _parse_cols(input_prefix: str, cols_raw: Any) -> tuple[tuple[str, ...], tuple[tuple[str, str], ...]]:
    """Normalize ``cols`` in either plain-list or aggregation-mapping form.

    Returns ``(cols, aggregations)`` — ``aggregations`` is empty for the
    list form, populated from the mapping form. Kept separate from
    :meth:`parse` so the type-dispatch logic is easy to read.
    """
    # Mapping form: ``cols: {deathtime: min, admittime: max}``. Aggregation-name
    # validation happens in ``__post_init__`` (validation-at-construction), so unknown
    # names still surface at parse time, not at the first ``group_by().agg()`` call
    # downstream.
    if isinstance(cols_raw, dict):
        for col in cols_raw:
            if not isinstance(col, str):
                raise ValueError(
                    f"Join config for '{input_prefix}' aggregation column names must be strings, "
                    f"got {type(col).__name__}: {col!r}."
                )
        return tuple(cols_raw), tuple(cols_raw.items())

    # A likely YAML slip: writing the aggregated form as a *list* of one-entry
    # mappings (``cols:`` / ``- deathtime: min``) instead of one mapping. YAML parses
    # that as ``[{"deathtime": "min"}]`` — catch it with a targeted message before the
    # generic list-form error below garbles the intent.
    if isinstance(cols_raw, list | tuple) and any(isinstance(c, dict) for c in cols_raw):
        raise ValueError(
            f"Join config for '{input_prefix}': 'cols' is a list containing mappings "
            f"({cols_raw!r}). For an aggregated join, write 'cols' as a single mapping — "
            f"'cols: {{deathtime: min}}' or an indented block WITHOUT '-' list markers — "
            f"not a YAML list of '- name: agg' items."
        )

    # Plain-list form: ``cols: [patient_id, dischtime]``. No aggregation.
    if not isinstance(cols_raw, list | tuple) or any(not isinstance(c, str) for c in cols_raw):
        raise ValueError(
            f"Join config for '{input_prefix}' must specify 'cols' as a list of column-name "
            f"strings (or a {{name: aggregation}} mapping for grouped joins), got "
            f"{type(cols_raw).__name__}: {cols_raw!r}. A bare string like "
            f"'cols: subject_id' would silently be treated as a tuple of characters — "
            f"use 'cols: [subject_id]' instead."
        )
    return tuple(cols_raw), ()

apply(left, input_dir)

Scan join-target files under input_dir and left-join them to left.

File resolution goes through :func:MEDS_extract.io.resolve_source_files, so every stage that applies a join uses the same layout-detection logic as the stages that read the main table. When aggregations is non-empty, the right-hand side is grouped by right_on and each named column is reduced before the join.

Every joined column must arrive under a name the left table does not already have: polars would deliver a colliding column under a _right-suffixed name, which MESSY expressions and the source-column plan cannot model, so the collision is rejected here with the offending names instead. (Same-named key columns are exempt — a left join coalesces them into one column.)

Examples:

End-to-end aggregated join — the admissions side has three rows for subject 1 (three admissions) and two rows for subject 2; the join pulls in the minimum deathtime per subject, fanning out over the patients table only once per subject:

>>> _ = pl.Config.set_tbl_width_chars(600)
>>> with yaml_disk('''
... hosp/admissions.parquet:
...   subject_id: [1, 1, 1, 2, 2]
...   deathtime:  ["2020-03-01", "2020-03-05", null, null, null]
... ''') as d:
...     jc = JoinConfig.parse({
...         "hosp/admissions": {"key": "subject_id", "cols": {"deathtime": "min"}}
...     })
...     patients = pl.LazyFrame({"subject_id": [1, 2, 3]})
...     jc.apply(patients, Path(d)).sort("subject_id").collect()
shape: (3, 2)
┌────────────┬────────────┐
│ subject_id ┆ deathtime  │
│ ---        ┆ ---        │
│ i64        ┆ str        │
╞════════════╪════════════╡
│ 1          ┆ 2020-03-01 │
│ 2          ┆ null       │
│ 3          ┆ null       │
└────────────┴────────────┘

A joined column colliding with a left column is rejected:

>>> with yaml_disk('''
... stays.parquet: {stay_id: [1], age: [40]}
... ''') as d:
...     jc = JoinConfig.parse({"stays": {"key": "stay_id", "cols": ["age"]}})
...     jc.apply(pl.LazyFrame({"stay_id": [1], "age": [39]}), Path(d))
Traceback (most recent call last):
    ...
ValueError: Join config for 'stays': joined column(s) ['age'] already exist on the left
table. Polars would deliver them under '_right'-suffixed names, which MESSY expressions
and the source-column plan cannot model, so plans and references would silently
mis-resolve. Joined columns must arrive under names the left table does not already
have: rename the column in the source data, or compute the value in a pre-processing
step.

Source code in MEDS_extract/config.py

def apply(self, left: pl.LazyFrame, input_dir: Path | UPath) -> pl.LazyFrame:
    """Scan join-target files under ``input_dir`` and left-join them to ``left``.

    File resolution goes through :func:`MEDS_extract.io.resolve_source_files`,
    so every stage that applies a join uses the same layout-detection logic
    as the stages that read the main table. When ``aggregations`` is
    non-empty, the right-hand side is grouped by ``right_on`` and each
    named column is reduced before the join.

    Every joined column must arrive under a name the left table does not
    already have: polars would deliver a colliding column under a
    ``_right``-suffixed name, which MESSY expressions and the source-column
    plan cannot model, so the collision is rejected here with the offending
    names instead. (Same-named key columns are exempt — a left join
    coalesces them into one column.)

    Examples:
        End-to-end aggregated join — the admissions side has three rows for
        subject 1 (three admissions) and two rows for subject 2; the join
        pulls in the minimum ``deathtime`` per subject, fanning out over the
        patients table only once per subject:

        >>> _ = pl.Config.set_tbl_width_chars(600)
        >>> with yaml_disk('''
        ... hosp/admissions.parquet:
        ...   subject_id: [1, 1, 1, 2, 2]
        ...   deathtime:  ["2020-03-01", "2020-03-05", null, null, null]
        ... ''') as d:
        ...     jc = JoinConfig.parse({
        ...         "hosp/admissions": {"key": "subject_id", "cols": {"deathtime": "min"}}
        ...     })
        ...     patients = pl.LazyFrame({"subject_id": [1, 2, 3]})
        ...     jc.apply(patients, Path(d)).sort("subject_id").collect()
        shape: (3, 2)
        ┌────────────┬────────────┐
        │ subject_id ┆ deathtime  │
        │ ---        ┆ ---        │
        │ i64        ┆ str        │
        ╞════════════╪════════════╡
        │ 1          ┆ 2020-03-01 │
        │ 2          ┆ null       │
        │ 3          ┆ null       │
        └────────────┴────────────┘

        A joined column colliding with a left column is rejected:

        >>> with yaml_disk('''
        ... stays.parquet: {stay_id: [1], age: [40]}
        ... ''') as d:
        ...     jc = JoinConfig.parse({"stays": {"key": "stay_id", "cols": ["age"]}})
        ...     jc.apply(pl.LazyFrame({"stay_id": [1], "age": [39]}), Path(d))
        Traceback (most recent call last):
            ...
        ValueError: Join config for 'stays': joined column(s) ['age'] already exist on the left
        table. Polars would deliver them under '_right'-suffixed names, which MESSY expressions
        and the source-column plan cannot model, so plans and references would silently
        mis-resolve. Joined columns must arrive under names the left table does not already
        have: rename the column in the source data, or compute the value in a pre-processing
        step.
    """
    left_schema = left.collect_schema().names()
    coalesced_keys = {r for lft, r in zip(self.left_on, self.right_on, strict=False) if lft == r}
    colliding = [c for c in self.cols if c in left_schema and c not in coalesced_keys]
    if colliding:
        raise ValueError(
            f"Join config for '{self.input_prefix}': joined column(s) {colliding} already "
            f"exist on the left table. Polars would deliver them under '_right'-suffixed "
            f"names, which MESSY expressions and the source-column plan cannot model, so "
            f"plans and references would silently mis-resolve. Joined columns must arrive "
            f"under names the left table does not already have: rename the column in the "
            f"source data, or compute the value in a pre-processing step."
        )
    right = scan_source(resolve_source_files(input_dir, self.input_prefix))
    if self.aggregations:
        right = self._aggregate(right)
    # ``maintain_order`` pins the join output to left-row order (ties broken by right
    # order). The in-memory engine happens to preserve left order anyway (this is a
    # no-op there), but the streaming engine — which convert_to_subject_sharded's
    # sink-based write runs on — reorders nondeterministically without it, and merge's
    # stable sort propagates that order into the final MEDS bytes for same-time events.
    return left.join(
        right,
        left_on=list(self.left_on),
        right_on=list(self.right_on),
        how="left",
        maintain_order="left_right",
    )

MessyConfig dataclass

The ONE top-level config class for a whole MESSY document.

A MESSY file carries up to three kinds of content — raw-data sources:, the etl: run block, and the event-conversion tables — and every entry point follows the same shape: take a spec, MessyConfig.load it once, then pull what it needs off the loaded object. Any section may be absent (a sources-only download spec, a download-free ETL, an etl:-free registered dataset); every section that IS present is validated at load, in full-document context, and accessors for absent sections raise clear errors at access time.

Attributes:

Name	Type	Description
source_fp	Path | None	The local file the spec resolved to (None for :meth:parse-built instances).
etl	EtlConfig	The parsed reserved etl: section (all-defaults when absent).
sources_version	str | dict[str, str] | None	The reserved sources.dataset_version value — scalar, per-bucket mapping, or None when absent.
spec_ref	str | None	The portable spec reference child processes should use — the pkg:// form for registered/pkg:// specs, else the absolute path.
registered_name	str | None	The MEDS_extract.pipelines entry-point name the spec resolved through, when it did (feeds :attr:dataset_name).
dist_version	str | None	The providing distribution’s version for registry-resolved specs (feeds :meth:dataset_version_for).
raw_doc	DictConfig | None	The raw, UNRESOLVED document. Kept in memory (not as a path: a :meth:parse-built instance has no file, and the accessors must agree with what was loaded even if the file changes on disk) for accessor-time resolution — interpolations stay symbolic until :meth:selected_sources selects a bucket / :attr:event_tables materializes the event section.
tables_raw	Mapping[str, Any] | DictConfig | None	The stripped, UNRESOLVED event-conversion section (tables + _defaults), materialized by :attr:event_tables on first access — resolution of this section is consumer-contextual exactly like the sources buckets’ (the download CLI must never need event-side ${oc.env:...} vars).

The surface, by consumer:

• Stages (MESSY_config_fp): :attr:event_tables and the table accessors built on it (:attr:table_prefixes, :meth:iter_tables, :meth:shuffled_tables, :meth:needed_source_columns, :meth:events_by_metadata_prefix), plus :meth:save.
• meds-extract-download: :meth:selected_sources — per-selected-bucket interpolation resolution happens here, at accessor time, against the raw unresolved document kept on the instance, so unselected buckets’ env vars are never required.
• meds-extract-run: :attr:etl (the parsed block), :attr:dataset_name, :meth:raw_version_for / :meth:dataset_version_for, and :meth:pipeline_config.

Examples:

>>> cfg = MessyConfig.parse({
...     "_defaults": {"subject_id": "$MRN"},
...     "patients": {"dob": {"code": "BIRTH", "time": "$dob"}},
...     "labs": {
...         "_defaults": {"subject_id": "$patient_id"},
...         "_table": {"join": {"stays": {"key": "stay_id", "cols": ["patient_id"]}}},
...         "lab": {"code": "$test", "time": "$ts"},
...     },
... })
>>> cfg.table_prefixes
['patients', 'labs']
>>> sorted(cfg.event_tables[0].subject_id_node.referenced_columns)
['MRN']
>>> [e.name for t in cfg.event_tables for e in t.events]
['dob', 'lab']

Source code in MEDS_extract/config.py

@dataclass(frozen=True)
class MessyConfig:
    """The ONE top-level config class for a whole MESSY document.

    A MESSY file carries up to three kinds of content — raw-data ``sources:``, the
    ``etl:`` run block, and the event-conversion tables — and every entry point
    follows the same shape: take a spec, ``MessyConfig.load`` it once, then pull
    what it needs off the loaded object. Any section may be absent (a sources-only
    download spec, a download-free ETL, an ``etl:``-free registered dataset); every
    section that IS present is validated at load, in full-document context, and
    accessors for absent sections raise clear errors at access time.

    Attributes:
        source_fp: The local file the spec resolved to (``None`` for
            :meth:`parse`-built instances).
        etl: The parsed reserved ``etl:`` section (all-defaults when absent).
        sources_version: The reserved ``sources.dataset_version`` value — scalar,
            per-bucket mapping, or ``None`` when absent.
        spec_ref: The portable spec reference child processes should use — the
            ``pkg://`` form for registered/``pkg://`` specs, else the absolute path.
        registered_name: The ``MEDS_extract.pipelines`` entry-point name the spec
            resolved through, when it did (feeds :attr:`dataset_name`).
        dist_version: The providing distribution's version for registry-resolved
            specs (feeds :meth:`dataset_version_for`).
        raw_doc: The raw, UNRESOLVED document. Kept in memory (not as a path: a
            :meth:`parse`-built instance has no file, and the accessors must agree
            with what was loaded even if the file changes on disk) for
            accessor-time resolution — interpolations stay symbolic until
            :meth:`selected_sources` selects a bucket / :attr:`event_tables`
            materializes the event section.
        tables_raw: The stripped, UNRESOLVED event-conversion section (tables +
            ``_defaults``), materialized by :attr:`event_tables` on first access —
            resolution of this section is consumer-contextual exactly like the
            sources buckets' (the download CLI must never need event-side
            ``${oc.env:...}`` vars).

    The surface, by consumer:

    - **Stages** (``MESSY_config_fp``): :attr:`event_tables` and the
      table accessors built on it (:attr:`table_prefixes`, :meth:`iter_tables`,
      :meth:`shuffled_tables`, :meth:`needed_source_columns`,
      :meth:`events_by_metadata_prefix`), plus :meth:`save`.
    - **`meds-extract-download`**: :meth:`selected_sources` — per-**selected**-bucket
      interpolation resolution happens here, at accessor time, against the raw
      unresolved document kept on the instance, so unselected buckets' env vars are
      never required.
    - **`meds-extract-run`**: :attr:`etl` (the parsed block), :attr:`dataset_name`,
      :meth:`raw_version_for` / :meth:`dataset_version_for`, and
      :meth:`pipeline_config`.

    Examples:
        >>> cfg = MessyConfig.parse({
        ...     "_defaults": {"subject_id": "$MRN"},
        ...     "patients": {"dob": {"code": "BIRTH", "time": "$dob"}},
        ...     "labs": {
        ...         "_defaults": {"subject_id": "$patient_id"},
        ...         "_table": {"join": {"stays": {"key": "stay_id", "cols": ["patient_id"]}}},
        ...         "lab": {"code": "$test", "time": "$ts"},
        ...     },
        ... })
        >>> cfg.table_prefixes
        ['patients', 'labs']
        >>> sorted(cfg.event_tables[0].subject_id_node.referenced_columns)
        ['MRN']
        >>> [e.name for t in cfg.event_tables for e in t.events]
        ['dob', 'lab']
    """

    source_fp: Path | None = None
    etl: EtlConfig = field(default_factory=EtlConfig)
    sources_version: str | dict[str, str] | None = None
    spec_ref: str | None = None
    registered_name: str | None = None
    dist_version: str | None = None
    raw_doc: DictConfig | None = field(default=None, repr=False, compare=False)
    tables_raw: Mapping[str, Any] | DictConfig | None = field(default=None, repr=False, compare=False)

    # The entry-point group dataset packages register under, mapping a public
    # dataset name directly to its bundled MESSY file ("<module>:<filename.yaml>").
    PIPELINES_ENTRY_POINT_GROUP: ClassVar[str] = "MEDS_extract.pipelines"

    # Top-level keys that are NOT event-table definitions. ``_defaults`` is consumed
    # separately below as the global defaults; ``sources`` and ``etl`` are the
    # reserved sibling sections parsed into their own fields.
    _RESERVED_TOP_LEVEL_KEYS: ClassVar[frozenset[str]] = frozenset({"sources", "etl"})

    # The subset of reserved keys that can carry credentials (literal API keys /
    # passwords in ``sources:`` backend configs) and therefore must ALSO be redacted
    # from log output (:meth:`load`) and from output-tree copies (:meth:`save`).
    # ``etl`` is deliberately NOT here: it carries only dataset name / version /
    # stage-option data, which is useful provenance in both places.
    _CREDENTIALED_TOP_LEVEL_KEYS: ClassVar[frozenset[str]] = frozenset({"sources"})

    @classmethod
    def _reject_table_shaped_reserved_block(cls, key: str, block: Any) -> None:
        """Reject a reserved top-level block whose content is event-table-shaped.

        ``sources`` and ``etl`` are reserved top-level names, stripped from the
        event-table remainder during :meth:`parse` — so a table configured under
        one (e.g. for a raw file named ``sources.csv``) would otherwise be
        silently dropped from the event plan, losing every event it defines. An
        event-table block is recognized by its reserved sub-keys (``_defaults``
        / ``_table``) or by any event-shaped entry (a mapping with a ``code``
        key); neither shape occurs in legitimate reserved blocks, whose bucket
        values are lists and whose settings are scalars.

        Examples:
            >>> MessyConfig.parse({
            ...     "sources": {"_defaults": {"subject_id": "$sid"}, "e": {"code": "X", "time": None}},
            ... })
            Traceback (most recent call last):
                ...
            ValueError: Top-level key 'sources' is reserved for download/ETL configuration, but its
            content looks like an event-table block (found '_defaults'; event-shaped entries ['e']).
            Tables cannot be named 'etl' or 'sources' — rename the source file, or nest it under a
            directory so the table prefix differs (e.g. 'raw/sources').

            A legitimate ``sources:`` block (buckets are lists, settings scalars) is untouched:

            >>> MessyConfig.parse({"sources": {"dataset": [], "dataset_version": "1.0"}}).sources_version
            '1.0'
        """
        if not isinstance(block, Mapping):
            return
        markers = sorted({"_defaults", "_table"} & set(block))
        event_like = sorted(k for k, v in block.items() if isinstance(v, Mapping) and "code" in v)
        if not markers and not event_like:
            return
        found = "; ".join(
            part
            for part in (
                ", ".join(f"'{m}'" for m in markers),
                f"event-shaped entries {event_like}" if event_like else "",
            )
            if part
        )
        names = " or ".join(f"'{k}'" for k in sorted(cls._RESERVED_TOP_LEVEL_KEYS))
        raise ValueError(
            f"Top-level key '{key}' is reserved for download/ETL configuration, but its content "
            f"looks like an event-table block (found {found}). Tables cannot be named {names} — "
            f"rename the source file, or nest it under a directory so the table prefix differs "
            f"(e.g. 'raw/{key}')."
        )

    @staticmethod
    def _sources_dataset_version(doc: DictConfig | Mapping | Any) -> str | dict[str, str] | None:
        """Read + shape-validate the reserved ``sources.dataset_version`` node of a spec document.

        Scalar string (one version for every bucket) or ``{bucket: version}`` mapping
        (demo/full releases genuinely differ). Being an ordinary document node it is
        interpolatable into source URLs (``${sources.dataset_version}``); only this node
        is resolved here. ``MessyConfig.save``'s sources redaction strips it from
        output-tree copies — fine, since the stamped value lands durably in
        ``metadata/dataset.json``.

        Examples:
            >>> doc = OmegaConf.create({"sources": {"dataset_version": "3.1"}})
            >>> MessyConfig._sources_dataset_version(doc)
            '3.1'
            >>> MessyConfig._sources_dataset_version(OmegaConf.create({"patients": {}})) is None
            True
            >>> MessyConfig._sources_dataset_version(OmegaConf.create({"sources": {"dataset_version": 3.1}}))
            Traceback (most recent call last):
                ...
            ValueError: sources.dataset_version must be a version string (quote it in YAML:
            dataset_version: "3.1") or a {bucket: version string} mapping, got float (3.1).
        """
        sources_node = doc.get("sources") if isinstance(doc, DictConfig | Mapping) else None
        if sources_node is None or "dataset_version" not in sources_node:
            return None
        # Scalar access through DictConfig resolves interpolations directly; a mapping
        # comes back as a DictConfig and needs an explicit resolving conversion.
        value = sources_node["dataset_version"]
        if OmegaConf.is_config(value):
            value = OmegaConf.to_container(value, resolve=True)
        ok = nonempty_str_err(value) is None or (
            isinstance(value, dict)
            and value
            and all(nonempty_str_err(b) is None and nonempty_str_err(v) is None for b, v in value.items())
        )
        if not ok:
            raise ValueError(
                f"sources.dataset_version must be a version string (quote it in YAML: "
                f'dataset_version: "{value}") or a {{bucket: version string}} mapping, got '
                f"{type(value).__name__} ({value!r})."
            )
        return value

    @classmethod
    def parse(cls, raw: Mapping[str, Any] | DictConfig) -> MessyConfig:
        """Parse a raw MESSY mapping into a :class:`MessyConfig`.

        Every present section is validated here; absent sections parse to their
        neutral field values (``tables=()``, all-defaults :attr:`etl`,
        ``sources_version=None``) and mis-ACCESS of an absent section raises later,
        at the accessor. The reserved sections are captured before the reserved
        keys are stripped, and stripping happens *before* interpolation resolution,
        so a download-only ``${oc.env:...}`` inside ``sources:`` never requires its
        env var to be set just to load the event-conversion side:

        >>> cfg = MessyConfig.parse(OmegaConf.create({
        ...     "sources": {"dataset": [{"type": "fsspec", "root": "${oc.env:UNSET_DOWNLOAD_ROOT}"}]},
        ...     "_defaults": {"subject_id": "$patient_id"},
        ...     "patients": {"dob": {"code": "DOB", "time": "$dob"}},
        ... }))
        >>> cfg.table_prefixes
        ['patients']

        The reserved sections land on their own fields, parsed in full-document
        context — config mistakes in them fail at load time in every consumer:

        >>> cfg = MessyConfig.parse({
        ...     "sources": {"dataset_version": "0.1"},
        ...     "etl": {"dataset_name": "Example"},
        ...     "patients": {"dob": {"code": "DOB", "time": "$dob"}},
        ... })
        >>> (cfg.etl.dataset_name, cfg.sources_version)
        ('Example', '0.1')
        >>> MessyConfig.parse({
        ...     "etl": {"pipeline": ["convert_to_parquet"]},
        ...     "patients": {"dob": {"code": "DOB", "time": "$dob"}},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: etl: block contains unknown key(s) ['pipeline']. ...

        A sources-only document is valid (it can drive ``meds-extract-download``);
        the event-conversion side errors at ACCESS, not load:

        >>> cfg = MessyConfig.parse({"sources": {"dataset": []}})
        >>> cfg.event_tables
        Traceback (most recent call last):
            ...
        ValueError: This MESSY spec declares no event tables ...

        Event-section mistakes surface at the first :attr:`event_tables` access —
        i.e. immediately after load in every event-conversion consumer, and never
        for the download CLI (whose spec may legitimately carry unresolvable
        event-side interpolations). ``_metadata`` mistakes, for example, name the
        event, the prefix, and the components the event offers:

        >>> cfg = MessyConfig.parse({
        ...     "chartevents": {
        ...         "chart": {
        ...             "code": 'f"CHART//{$itemid}"',
        ...             "time": None,
        ...             "_metadata": {"d_items": {"description": "$label"}},
        ...         },
        ...     },
        ... })
        >>> cfg.event_tables
        Traceback (most recent call last):
            ...
        ValueError: _metadata block (event 'chart', metadata prefix 'd_items') produces no join-key
        columns: ... Component columns available on this event: ['itemid'] ...
        """
        # Keep the pristine, UNRESOLVED document for accessor-time resolution
        # (``selected_sources`` / ``event_tables``); everything below works on copies.
        raw_doc = OmegaConf.create(raw)
        for key in cls._RESERVED_TOP_LEVEL_KEYS:
            if key in raw_doc:
                cls._reject_table_shaped_reserved_block(
                    key, OmegaConf.to_container(raw_doc[key], resolve=False)
                )
        sources_version = cls._sources_dataset_version(raw)
        etl_raw = OmegaConf.to_container(raw_doc.etl, resolve=False) if "etl" in raw_doc else None

        # The event-conversion remainder: reserved keys stripped, NOT resolved —
        # its interpolations resolve at ``event_tables`` access, so a download-only
        # consumer never needs event-side env vars and vice versa.
        stripped = OmegaConf.create(raw_doc)
        for key in cls._RESERVED_TOP_LEVEL_KEYS:
            if key in stripped:
                del stripped[key]

        return cls(
            etl=EtlConfig.parse(etl_raw),
            sources_version=sources_version,
            raw_doc=raw_doc,
            tables_raw=stripped,
        )

    @classmethod
    def load(cls, spec: str | Path) -> MessyConfig:
        """THE loading entry point: resolve a spec reference, read, validate, parse.

        Every consumer — the run CLI, the download CLI, and all eight stages (via
        ``MESSY_config_fp``) — funnels through this one call. ``spec``
        resolves down the same ladder ``MEDS_transform-pipeline`` has for pipeline
        configs, extended one rung up with the registry:

        1. **Registered name** — exact match against the ``MEDS_extract.pipelines``
           entry-point group. The registration points directly at the bundled file
           (``"<package.module>:<filename.yaml>"``); its raw ``.value`` string is
           parsed here — never ``load()``-ed, so registration cannot execute
           dataset-package code. The registered name and the providing
           distribution's version land on :attr:`registered_name` /
           :attr:`dist_version` (feeding :attr:`dataset_name` and
           :meth:`dataset_version_for`), and :attr:`spec_ref` becomes the
           equivalent portable ``pkg://`` reference.
        2. **pkg://** — resolved via the shared :func:`resolve_config_path`.
        3. **Explicit filesystem path** — an absolute path, a ``~`` path, or an
           explicit relative path (``./x.yaml`` / ``../x.yaml``). A bare name that
           matches no registration is an error naming both remedies (the registered
           names, and the ``./`` spelling) — it is NOT tried as a relative path, so
           a typo'd dataset name can never silently resolve to a stray local file,
           and an explicit path never collides with an entry-point name.

        Examples:
            >>> with yaml_disk('''
            ... spec.yaml: |
            ...   etl: {dataset_name: Example, raw_dataset_version: "0.1"}
            ...   patients:
            ...     dob: {code: BIRTH, time: null}
            ... ''') as d:
            ...     messy = MessyConfig.load(Path(d) / "spec.yaml")
            ...     (messy.dataset_name, messy.raw_version_for(), messy.spec_ref == str(messy.source_fp))
            ('Example', '0.1', True)

            The registry rung, demonstrated against a synthetic installed package +
            registrations (the ``fake_pipeline_registry`` test fixture): the
            registered name supplies :attr:`dataset_name`, the distribution supplies
            the version-stamp suffix, and :attr:`spec_ref` is the portable ``pkg://``
            form:

            >>> getfixture("fake_pipeline_registry")
            >>> messy = MessyConfig.load("Fake-DS")
            >>> (messy.dataset_name, messy.spec_ref, messy.dataset_version_for())
            ('Fake-DS', 'pkg://fake_ds_pkg.event_configs.yaml', '0.9:1.2.3')

            Malformed registrations fail with targeted messages — a bare module
            reference (no ``:filename``) and a registration naming a resource the
            module doesn't bundle:

            >>> MessyConfig.load("Bare-Mod")
            Traceback (most recent call last):
                ...
            ValueError: Entry point 'Bare-Mod' in group 'MEDS_extract.pipelines' has value
            'fake_ds_pkg', which does not name the bundled MESSY file. Register it as
            '<package.module>:<filename.yaml>', ...
            >>> MessyConfig.load("Missing-Res")
            Traceback (most recent call last):
                ...
            ValueError: Entry point 'Missing-Res' points at 'fake_ds_pkg:nope.yaml', but module
            'fake_ds_pkg' has no resource named 'nope.yaml' ...

            A bare name matching no registration fails with both remedies — the
            registered names, and the explicit-path spelling (bare relative paths are
            deliberately not tried, so ``file.yaml`` errors while ``./file.yaml``
            loads):

            >>> MessyConfig.load("Not-A-Registered-Name")
            Traceback (most recent call last):
                ...
            FileNotFoundError: spec='Not-A-Registered-Name' is not a registered pipeline name or a
            pkg:// reference. Registered pipelines: Bare-Mod, Fake-DS, Missing-Res. To load a MESSY
            file by path, give an absolute path or an explicit relative path
            (./Not-A-Registered-Name).

            An explicit path that does not exist says so directly:

            >>> MessyConfig.load("./no-such-file.yaml")  # doctest: +ELLIPSIS
            Traceback (most recent call last):
                ...
            FileNotFoundError: spec='./no-such-file.yaml' resolved to ...no-such-file.yaml, which
            does not exist.
        """
        spec = str(spec)
        by_name = {ep.name: ep for ep in entry_points(group=cls.PIPELINES_ENTRY_POINT_GROUP)}
        registered_name = dist_version = None
        if spec in by_name:
            ep = by_name[spec]
            module, _, filename = ep.value.partition(":")
            if not module or not filename:
                raise ValueError(
                    f"Entry point {spec!r} in group {cls.PIPELINES_ENTRY_POINT_GROUP!r} has value "
                    f"{ep.value!r}, which does not name the bundled MESSY file. Register it as "
                    f"'<package.module>:<filename.yaml>', e.g. "
                    f"'MIMIC_IV_MEDS.configs:event_configs.yaml'."
                )
            spec_ref = f"{PKG_PFX}{module}.{filename.replace('/', '.')}"
            fp = Path(str(files(module) / filename))
            if not fp.is_file():
                raise ValueError(
                    f"Entry point {spec!r} points at {ep.value!r}, but module {module!r} has no "
                    f"resource named {filename!r} (resolved to {fp})."
                )
            registered_name = ep.name
            dist_version = ep.dist.version if ep.dist is not None else None
        elif spec.startswith(PKG_PFX):
            spec_ref = spec
            fp = resolve_config_path(spec)
            if not fp.is_file():
                raise FileNotFoundError(f"spec={spec!r} resolved to {fp}, which does not exist.")
        elif spec.startswith(("./", "../", "~")) or Path(spec).is_absolute():
            fp = Path(spec).expanduser().resolve()
            if not fp.is_file():
                raise FileNotFoundError(f"spec={spec!r} resolved to {fp}, which does not exist.")
            spec_ref = str(fp)
        else:
            # A bare name is ONLY a registry lookup: never fall back to treating it as
            # a relative path, so a typo'd dataset name cannot silently resolve to a
            # stray local file (and an explicit path never collides with a name).
            registered = ", ".join(sorted(by_name)) or "(none)"
            raise FileNotFoundError(
                f"spec={spec!r} is not a registered pipeline name or a pkg:// reference. "
                f"Registered pipelines: {registered}. To load a MESSY file by path, give an "
                f"absolute path or an explicit relative path (./{spec})."
            )

        logger.info(f"Reading MESSY config from {fp}")
        raw = OmegaConf.load(fp)
        # Log with credential-bearing reserved keys stripped: a combined-MESSY
        # ``sources:`` block can carry credentials (literal API keys / passwords),
        # which must not land in every stage's log output. The ``etl:`` block is
        # credential-free and stays in the log — it's useful provenance.
        loggable = OmegaConf.create(raw)
        for key in cls._CREDENTIALED_TOP_LEVEL_KEYS:
            if key in loggable:
                del loggable[key]
        logger.info(f"MESSY config:\n{OmegaConf.to_yaml(loggable)}")
        parsed = cls.parse(raw)
        # Attach the spec context (frozen dataclass => object.__setattr__, the same
        # idiom the pre-existing source_fp attachment used).
        for attr, value in [
            ("source_fp", fp),
            ("spec_ref", spec_ref),
            ("registered_name", registered_name),
            ("dist_version", dist_version),
        ]:
            object.__setattr__(parsed, attr, value)
        return parsed

    # ── Section accessors ────────────────────────────────────────────

    @cached_property
    def event_tables(self) -> tuple[TableConfig, ...]:
        """The event-conversion tables, materialized (resolved + parsed) on first access.

        Every stage-facing accessor routes through this, so a sources-only spec
        handed to the event-conversion pipeline fails with the real cause named
        instead of silently no-op'ing through the first stage. Materialization is
        lazy for the same reason bucket resolution is: this section's
        ``${oc.env:...}`` interpolations belong to the event-conversion consumers,
        and the download CLI must never need them. Every event-consuming entry
        point accesses this immediately after load, so validation timing there is
        unchanged.

        Examples:
            >>> MessyConfig.parse({"sources": {"dataset": []}}).event_tables
            Traceback (most recent call last):
                ...
            ValueError: This MESSY spec declares no event tables (only reserved sections). A
            sources-only spec can drive `meds-extract-download`, but the event-conversion pipeline
            needs event-table definitions.

            The top-level ``_defaults`` block accepts only the keys it actually
            consumes — a stray key would otherwise be silently dropped:

            >>> MessyConfig.parse({
            ...     "_defaults": {"subject_id": "$MRN", "time": "$charttime"},
            ...     "patients": {"dob": {"code": "BIRTH", "time": "$dob"}},
            ... }).event_tables
            Traceback (most recent call last):
                ...
            ValueError: Top-level '_defaults' has unknown keys: ['time']. Allowed keys:
            'subject_id'.

            And an underscore-prefixed top-level key other than ``_defaults`` is
            almost certainly a typo of it, not a table prefix:

            >>> MessyConfig.parse({
            ...     "_default": {"subject_id": "$MRN"},
            ...     "patients": {"dob": {"code": "BIRTH", "time": "$dob"}},
            ... }).event_tables
            Traceback (most recent call last):
                ...
            ValueError: Unknown reserved key(s) ['_default'] at the top level of the MESSY event
            section. The only reserved key at this level is '_defaults'. Table prefixes may not
            begin with an underscore.
        """
        raw = self.tables_raw
        if OmegaConf.is_config(raw):
            raw = OmegaConf.to_container(raw, resolve=True)
        raw_dict = dict(raw or {})
        global_defaults = dict(raw_dict.pop("_defaults", {}))
        unknown_default_keys = set(global_defaults) - {"subject_id"}
        if unknown_default_keys:
            raise ValueError(
                f"Top-level '_defaults' has unknown keys: {sorted(unknown_default_keys)}. "
                f"Allowed keys: 'subject_id'."
            )
        stray_reserved = sorted(k for k in raw_dict if k.startswith("_"))
        if stray_reserved:
            raise ValueError(
                f"Unknown reserved key(s) {stray_reserved} at the top level of the MESSY event "
                f"section. The only reserved key at this level is '_defaults'. Table prefixes "
                f"may not begin with an underscore."
            )
        if not raw_dict:
            raise ValueError(
                "This MESSY spec declares no event tables (only reserved sections). A "
                "sources-only spec can drive `meds-extract-download`, but the event-conversion "
                "pipeline needs event-table definitions."
            )
        return tuple(TableConfig.parse(prefix, block, global_defaults) for prefix, block in raw_dict.items())

    def selected_sources(self, key: str = "dataset") -> list:
        """Construct the ``Source`` objects for one download run (bucket ``key`` + ``common``).

        Resolution is per-**selected**-bucket and happens here, at accessor time,
        against the raw unresolved document kept on the instance: only the ``key``
        bucket and the always-appended ``common`` bucket are resolved, each while
        still ATTACHED to the document (so document-relative references like
        ``${sources.dataset_version}`` keep resolving) — unselected buckets'
        ``${oc.env:...}`` credentials are never required.

        Returns ``[]`` for a document with no ``sources:`` block — deliberately not
        an access-time error, because "nothing to download" is a *legal state with
        defined semantics* for this accessor's one consumer (a download-free ETL
        warns and exits 0), unlike a table-less pipeline run. A ``key`` naming no
        bucket — including the reserved ``dataset_version`` metadata key — IS a
        config error (likely a typo): because ``common`` is always appended, a
        typo'd key would otherwise quietly fetch only the common bucket and
        "succeed".

        Examples:
            >>> with yaml_disk('''
            ... mirror:
            ...   patients.csv: "patient_id\\\\n1\\\\n"
            ... ''') as d:
            ...     spec_fp = Path(d) / "spec.yaml"
            ...     src_yaml = f"sources:\\n  dataset:\\n    - type: fsspec\\n      root: {d}/mirror\\n"
            ...     _ = spec_fp.write_text(src_yaml)
            ...     [type(s).__name__ for s in MessyConfig.load(spec_fp).selected_sources()]
            ['FsspecSource']

            >>> MessyConfig.parse({"patients": {"dob": {"code": "BIRTH", "time": None}}}
            ...     ).selected_sources()
            []

            >>> MessyConfig.parse({"sources": {"dataset": []}}).selected_sources(key="dataste")
            Traceback (most recent call last):
                ...
            ValueError: key='dataste' does not name a sources bucket. Available buckets:
            ['dataset'].
        """
        # Deferred: the download layer imports this module already (the reverse
        # import at module scope would be circular).
        from .download.spec import SOURCES_RESERVED_KEYS, sources_from_spec

        sources_node = self.raw_doc.get("sources") if self.raw_doc is not None else None
        # Bucket names come from the UNRESOLVED node — listing them must not require
        # any interpolation (in any bucket) to be resolvable.
        if sources_node and (key in SOURCES_RESERVED_KEYS or key not in sources_node):
            raise ValueError(
                f"key={key!r} does not name a sources bucket. "
                f"Available buckets: {sorted(set(sources_node) - SOURCES_RESERVED_KEYS)}."
            )
        sources_dict = {}
        if sources_node is not None:
            for bucket in dict.fromkeys((key, "common")):  # de-dupe when key="common"
                bucket_node = sources_node.get(bucket)
                if bucket_node is not None:
                    sources_dict[bucket] = OmegaConf.to_container(bucket_node, resolve=True)
        return sources_from_spec({"sources": sources_dict}, key=key)

    @property
    def dataset_name(self) -> str:
        """The dataset's name: ``etl.dataset_name``, else the registered spec name.

        Examples:
            >>> MessyConfig(etl=EtlConfig(dataset_name="X")).dataset_name
            'X'
            >>> MessyConfig(registered_name="MIMIC-IV").dataset_name
            'MIMIC-IV'
            >>> MessyConfig(source_fp=Path("/specs/messy.yaml")).dataset_name
            Traceback (most recent call last):
                ...
            ValueError: The etl: block in /specs/messy.yaml omits dataset_name, which is only
            allowed when the spec is resolved via a registered MEDS_extract.pipelines entry-point
            name (the dataset name then defaults to that name). For pkg://- and path-resolved
            specs, add dataset_name to the etl: block.
        """
        name = self.etl.dataset_name or self.registered_name
        if name is None:
            raise ValueError(
                f"The etl: block in {self.source_fp} omits dataset_name, which is only allowed "
                f"when the spec is resolved via a registered MEDS_extract.pipelines entry-point "
                f"name (the dataset name then defaults to that name). For pkg://- and "
                f"path-resolved specs, add dataset_name to the etl: block."
            )
        return name

    def raw_version_for(self, key: str = "dataset") -> str:
        """The effective raw-data version for one run (selected bucket ``key``).

        ``sources.dataset_version`` (scalar, or the mapping's entry for ``key``) is
        authoritative when present; ``etl.raw_dataset_version`` is the fallback.
        When both resolve they must match — silent divergence would stamp a lie
        into ``dataset.json``.

        Examples:
            >>> MessyConfig(sources_version={"dataset": "3.1", "demo": "2.2"}).raw_version_for("demo")
            '2.2'
            >>> MessyConfig(etl=EtlConfig(raw_dataset_version="0.1")).raw_version_for()
            '0.1'
            >>> MessyConfig(sources_version="3.1", etl=EtlConfig(raw_dataset_version="9.9")
            ...     ).raw_version_for()
            Traceback (most recent call last):
                ...
            ValueError: sources.dataset_version resolves to '3.1' for key='dataset' but
            etl.raw_dataset_version says '9.9'. These must match — keep one source of truth
            (prefer sources.dataset_version and drop the etl: fallback).
            >>> MessyConfig().raw_version_for()
            Traceback (most recent call last):
                ...
            ValueError: No raw dataset version declared: add `dataset_version` to the sources:
            block (scalar or per-bucket mapping), or `raw_dataset_version` to the etl: block.
        """
        from_sources = (
            self.sources_version.get(key)
            if isinstance(self.sources_version, Mapping)
            else self.sources_version
        )
        if from_sources and self.etl.raw_dataset_version and from_sources != self.etl.raw_dataset_version:
            raise ValueError(
                f"sources.dataset_version resolves to {from_sources!r} for key={key!r} but "
                f"etl.raw_dataset_version says {self.etl.raw_dataset_version!r}. These must "
                f"match — keep one source of truth (prefer sources.dataset_version and drop "
                f"the etl: fallback)."
            )
        effective = from_sources or self.etl.raw_dataset_version
        if effective is None:
            raise ValueError(
                "No raw dataset version declared: add `dataset_version` to the sources: block "
                "(scalar or per-bucket mapping), or `raw_dataset_version` to the etl: block."
            )
        return effective

    def dataset_version_for(self, key: str = "dataset", override: str | None = None) -> str:
        """The ``etl_metadata.dataset_version`` stamp for one run.

        ``{raw version}:{providing distribution's version}`` for registry-resolved
        specs; the raw version alone otherwise; ``override`` (the CLI's
        ``dataset_version=``) always wins.

        Examples:
            >>> MessyConfig(sources_version="3.1", dist_version="1.2.3").dataset_version_for()
            '3.1:1.2.3'
            >>> MessyConfig(sources_version="3.1").dataset_version_for(override="custom")
            'custom'
        """
        if override:
            return override
        raw = self.raw_version_for(key)
        return f"{raw}:{self.dist_version}" if self.dist_version else raw

    def pipeline_config(
        self,
        *,
        input_dir: Path,
        output_dir: Path,
        key: str = "dataset",
        dataset_version: str | None = None,
    ) -> dict[str, Any]:
        """The full MEDS-transforms pipeline config for one run, as a plain dict.

        Only meaningful on :meth:`load`-built instances (it needs the spec
        context). Every value is **inlined as a resolved literal** — no env-var
        indirection — so the written file is self-contained, diffable provenance
        and the only channel through which the computed identity reaches the
        ``MEDS_transform-pipeline`` subprocess. ``MESSY_config_fp``
        carries :attr:`spec_ref` (the portable ``pkg://`` form for registered/pkg
        specs), which every consumer resolves via this class's ladder-aware
        :meth:`load`.

        Examples:
            >>> messy = MessyConfig(
            ...     etl=EtlConfig.parse({"n_subjects_per_shard": 1000}),
            ...     sources_version="3.1",
            ...     dist_version="1.0.0",
            ...     registered_name="Example",
            ...     spec_ref="pkg://example_pkg.messy.yaml",
            ... )
            >>> cfg = messy.pipeline_config(
            ...     input_dir=Path("/data/raw_input"), output_dir=Path("/data/MEDS_cohort")
            ... )
            >>> print(OmegaConf.to_yaml(OmegaConf.create(cfg)).strip())
            etl_metadata:
              dataset_name: Example
              dataset_version: 3.1:1.0.0
            MESSY_config_fp: pkg://example_pkg.messy.yaml
            input_dir: /data/raw_input
            output_dir: /data/MEDS_cohort
            shards_map_fp: /data/MEDS_cohort/metadata/.shards.json
            stages:
            - convert_to_parquet
            - split_and_shard_subjects:
                n_subjects_per_shard: 1000
            - convert_to_subject_sharded
            - convert_to_MEDS_events
            - extract_code_metadata
            - merge_to_MEDS_cohort
            - finalize_MEDS_metadata
            - finalize_MEDS_data
        """
        return {
            "etl_metadata": {
                "dataset_name": self.dataset_name,
                "dataset_version": self.dataset_version_for(key, override=dataset_version),
            },
            "MESSY_config_fp": self.spec_ref,
            "input_dir": str(input_dir),
            "output_dir": str(output_dir),
            "shards_map_fp": f"{output_dir}/metadata/.shards.json",
            "stages": self.etl.stages_container(),
        }

    def save(self, fp: Path | UPath | str) -> None:
        """Copy the original MESSY config file to ``fp``, minus credentialed keys.

        Only valid on instances produced by :meth:`load` (which remembers the
        source path). Instances built via :meth:`parse` directly don't have a
        source file to copy and will raise. Uses ``read_bytes`` / ``write_bytes``
        so UPath-backed cloud destinations work as well as local paths.

        When the source file carries credential-bearing reserved blocks
        (``sources:``), the copy is re-serialized with those blocks stripped — a
        combined-MESSY ``sources:`` block can carry credentials, and this copy lands
        inside the (often shared) pipeline output tree. The credential-free ``etl:``
        block is NOT stripped: the stage list and dataset name/version are useful
        provenance in the output copy. Comment formatting is preserved only for
        files with no credentialed blocks, where a verbatim byte-copy suffices.

        Examples:
            >>> yaml = '''
            ... sources:
            ...   dataset:
            ...     - type: http
            ...       headers: {X-Dataverse-key: super-secret-token}
            ...       urls: [https://example.com/x.csv]
            ... etl:
            ...   dataset_name: Example
            ...   raw_dataset_version: "0.1"
            ... patients:
            ...   dob: {code: BIRTH, time: null}
            ... '''
            >>> cfg_fp = getfixture("tmp_path") / "cfg.yaml"
            >>> _ = cfg_fp.write_text(yaml)
            >>> out_fp = getfixture("tmp_path") / "copy.yaml"
            >>> MessyConfig.load(cfg_fp).save(out_fp)
            >>> print(out_fp.read_text().strip())
            etl:
              dataset_name: Example
              raw_dataset_version: '0.1'
            patients:
              dob:
                code: BIRTH
                time: null
            >>> "super-secret-token" in out_fp.read_text()
            False
        """
        if self.source_fp is None:
            raise ValueError("MessyConfig.save requires a source file path (only available after .load()).")
        if not self.source_fp.exists():
            raise FileNotFoundError(
                f"MessyConfig source file no longer exists at {self.source_fp}; cannot copy to {fp}."
            )
        dest = Path(fp) if isinstance(fp, str) else fp
        raw = OmegaConf.load(self.source_fp)
        credentialed_present = [k for k in self._CREDENTIALED_TOP_LEVEL_KEYS if k in raw]
        if not credentialed_present:
            dest.write_bytes(self.source_fp.read_bytes())
            return
        for key in credentialed_present:
            del raw[key]
        # ``to_yaml`` does not resolve interpolations, so symbolic ``${oc.env:...}``
        # references in the event-conversion sections survive the round-trip.
        dest.write_bytes(OmegaConf.to_yaml(raw).encode("utf-8"))

    def iter_tables(self) -> Iterator[TableConfig]:
        return iter(self.event_tables)

    def shuffled_tables(self) -> list[TableConfig]:
        """Return tables in randomized order.

        Used by stages that iterate tables to spread parallel worker load — without shuffling, every worker
        would contend on the same first table.
        """
        tables = list(self.event_tables)
        random.Random().shuffle(tables)
        return tables

    @property
    def table_prefixes(self) -> list[str]:
        return [t.input_prefix for t in self.event_tables]

    def needed_source_columns(self) -> dict[str, list[str]]:
        """Map each source prefix to the sorted list of columns that must be read.

        Aggregates each table's own :meth:`TableConfig.source_columns` plus the
        columns that any table pulls in from a join target. The returned dict is
        the input to ``convert_to_parquet`` — it tells that stage which columns
        to project when normalizing raw inputs.

        Examples:
            >>> cfg = MessyConfig.parse({
            ...     "_defaults": {"subject_id": "$subject_id_global"},
            ...     "hosp/patients": {
            ...         "eye_color": {"code": "EYE_COLOR", "time": None},
            ...         "height": {"code": "HEIGHT", "time": None, "numeric_value": "$height"},
            ...     },
            ...     "icu/chartevents": {
            ...         "_defaults": {"subject_id": "$subject_id_icu"},
            ...         "heart_rate": {
            ...             "code": "HEART_RATE", "time": "$charttime", "numeric_value": "$HR"
            ...         },
            ...     },
            ... })
            >>> cfg.needed_source_columns()
            {'hosp/patients': ['height', 'subject_id_global'],
             'icu/chartevents': ['HR', 'charttime', 'subject_id_icu']}

            Derived columns and joined columns are excluded from the source file's
            needed-column list (they come from elsewhere); the join target gets
            its own entry:

            >>> cfg = MessyConfig.parse({
            ...     "labs": {
            ...         "_defaults": {"subject_id": "$patient_id"},
            ...         "_table": {"join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}}},
            ...         "lab": {"code": "$test", "time": "$dischtime"},
            ...     },
            ... })
            >>> cfg.needed_source_columns()
            {'labs': ['patient_id', 'stay_id', 'test'], 'stays': ['dischtime', 'stay_id']}

            Aggregated joins plan the same way: the aggregation's
            source column is needed on the *right*-side table even though the
            left-side table never reads it directly — ``deathtime`` below is the
            input to ``min()`` on the admissions side:

            >>> cfg = MessyConfig.parse({
            ...     "hosp/patients": {
            ...         "_table": {
            ...             "join": {
            ...                 "hosp/admissions": {"key": "subject_id", "cols": {"deathtime": "min"}}
            ...             }
            ...         },
            ...         "death": {"code": "MEDS_DEATH", "time": "$deathtime"},
            ...     },
            ... })
            >>> cfg.needed_source_columns()
            {'hosp/patients': ['subject_id'], 'hosp/admissions': ['deathtime', 'subject_id']}

            Transform *outputs* are computed at read time, not read from disk, so they are
            excluded from the plan while their input columns are included.
            Format-annotated time strings contribute their source column, and ``_metadata``
            blocks contribute nothing:

            >>> cfg = MessyConfig.parse({
            ...     "hosp/patients": {
            ...         "_table": {"cols": {"year_of_birth": "$anchor_year - $anchor_age"}},
            ...         "dob": {"code": "MEDS_BIRTH", "time": "$year_of_birth::year"},
            ...         "admit": {
            ...             "code": 'f"ADMIT//{$admtype}"',
            ...             "time": '$admittime::"%Y-%m-%d %H:%M:%S"',
            ...             "_metadata": {
            ...                 "admissions_meta": {"admtype": "$admtype", "description": "$adm_desc"}
            ...             },
            ...         },
            ...     },
            ... })
            >>> cfg.needed_source_columns()
            {'hosp/patients': ['admittime', 'admtype', 'anchor_age', 'anchor_year', 'subject_id']}
        """
        out: dict[str, set[str]] = {}
        for table in self.event_tables:
            out.setdefault(table.input_prefix, set()).update(table.source_columns())
            if table.join is not None:
                jt = out.setdefault(table.join.input_prefix, set())
                jt.update(table.join.right_on)
                jt.update(table.join.cols)
        return {k: sorted(v) for k, v in out.items()}

    def events_by_metadata_prefix(self) -> dict[str, list[dict]]:
        """Invert the events → metadata mapping.

        Each event's ``_metadata`` block maps metadata-file prefixes to
        per-prefix metadata config dicts. This returns the reverse: each
        metadata prefix gets the list of ``{code, _metadata, source_block}``
        entries that reference it. The ``code`` value is always the original
        raw dftly expression string — a metadata-carrying event is guaranteed
        to retain it (enforced at :class:`EventConfig` construction), and it
        is what downstream ``code_template`` columns stamp verbatim. The
        ``source_block`` value is the
        ``{input_prefix}/{event_name}`` tag that :meth:`EventConfig.extract`
        stamps on every output row — ``extract_code_metadata`` uses it to
        scope metadata joins to the event that declared the ``_metadata``
        block.

        Used by ``extract_code_metadata``.

        Examples:
            >>> cfg = MessyConfig.parse({
            ...     "_defaults": {"subject_id": "$MRN"},
            ...     "icu/procedureevents": {
            ...         "_defaults": {"subject_id": "$subject_id"},
            ...         "start": {
            ...             "code": 'f"PROC//START//{$itemid}"',
            ...             "_metadata": {
            ...                 "proc_datetimeevents": {
            ...                     "itemid": "$itemid",
            ...                     "desc": "coalesce($omop_concept_name, $label)",
            ...                 },
            ...             },
            ...         },
            ...     },
            ... })
            >>> grouped = cfg.events_by_metadata_prefix()
            >>> sorted(grouped.keys())
            ['proc_datetimeevents']
            >>> entry = grouped["proc_datetimeevents"][0]
            >>> entry["code"]
            'f"PROC//START//{$itemid}"'
            >>> entry["source_block"]
            'icu/procedureevents/start'
            >>> MessyConfig.parse({"t": {"e": {"code": "X", "time": None}}}).events_by_metadata_prefix()
            {}
        """
        out: dict[str, list[dict]] = {}
        for table in self.event_tables:
            for event in table.events:
                source_block = f"{table.input_prefix}/{event.name}"
                for metadata_prefix, metadata_cfg in event.metadata.items():
                    out.setdefault(metadata_prefix, []).append(
                        {"code": event.raw_code, "_metadata": metadata_cfg, SOURCE_BLOCK_COL: source_block}
                    )
        return out

dataset_name property

The dataset’s name: etl.dataset_name, else the registered spec name.

Examples:

>>> MessyConfig(etl=EtlConfig(dataset_name="X")).dataset_name
'X'
>>> MessyConfig(registered_name="MIMIC-IV").dataset_name
'MIMIC-IV'
>>> MessyConfig(source_fp=Path("/specs/messy.yaml")).dataset_name
Traceback (most recent call last):
    ...
ValueError: The etl: block in /specs/messy.yaml omits dataset_name, which is only
allowed when the spec is resolved via a registered MEDS_extract.pipelines entry-point
name (the dataset name then defaults to that name). For pkg://- and path-resolved
specs, add dataset_name to the etl: block.

event_tables cached property

The event-conversion tables, materialized (resolved + parsed) on first access.

Every stage-facing accessor routes through this, so a sources-only spec handed to the event-conversion pipeline fails with the real cause named instead of silently no-op’ing through the first stage. Materialization is lazy for the same reason bucket resolution is: this section’s ${oc.env:...} interpolations belong to the event-conversion consumers, and the download CLI must never need them. Every event-consuming entry point accesses this immediately after load, so validation timing there is unchanged.

Examples:

>>> MessyConfig.parse({"sources": {"dataset": []}}).event_tables
Traceback (most recent call last):
    ...
ValueError: This MESSY spec declares no event tables (only reserved sections). A
sources-only spec can drive `meds-extract-download`, but the event-conversion pipeline
needs event-table definitions.

The top-level _defaults block accepts only the keys it actually consumes — a stray key would otherwise be silently dropped:

>>> MessyConfig.parse({
...     "_defaults": {"subject_id": "$MRN", "time": "$charttime"},
...     "patients": {"dob": {"code": "BIRTH", "time": "$dob"}},
... }).event_tables
Traceback (most recent call last):
    ...
ValueError: Top-level '_defaults' has unknown keys: ['time']. Allowed keys:
'subject_id'.

And an underscore-prefixed top-level key other than _defaults is almost certainly a typo of it, not a table prefix:

>>> MessyConfig.parse({
...     "_default": {"subject_id": "$MRN"},
...     "patients": {"dob": {"code": "BIRTH", "time": "$dob"}},
... }).event_tables
Traceback (most recent call last):
    ...
ValueError: Unknown reserved key(s) ['_default'] at the top level of the MESSY event
section. The only reserved key at this level is '_defaults'. Table prefixes may not
begin with an underscore.

_reject_table_shaped_reserved_block(key, block) classmethod

Reject a reserved top-level block whose content is event-table-shaped.

sources and etl are reserved top-level names, stripped from the event-table remainder during :meth:parse — so a table configured under one (e.g. for a raw file named sources.csv) would otherwise be silently dropped from the event plan, losing every event it defines. An event-table block is recognized by its reserved sub-keys (_defaults / _table) or by any event-shaped entry (a mapping with a code key); neither shape occurs in legitimate reserved blocks, whose bucket values are lists and whose settings are scalars.

Examples:

>>> MessyConfig.parse({
...     "sources": {"_defaults": {"subject_id": "$sid"}, "e": {"code": "X", "time": None}},
... })
Traceback (most recent call last):
    ...
ValueError: Top-level key 'sources' is reserved for download/ETL configuration, but its
content looks like an event-table block (found '_defaults'; event-shaped entries ['e']).
Tables cannot be named 'etl' or 'sources' — rename the source file, or nest it under a
directory so the table prefix differs (e.g. 'raw/sources').

A legitimate sources: block (buckets are lists, settings scalars) is untouched:

>>> MessyConfig.parse({"sources": {"dataset": [], "dataset_version": "1.0"}}).sources_version
'1.0'

Source code in MEDS_extract/config.py

@classmethod
def _reject_table_shaped_reserved_block(cls, key: str, block: Any) -> None:
    """Reject a reserved top-level block whose content is event-table-shaped.

    ``sources`` and ``etl`` are reserved top-level names, stripped from the
    event-table remainder during :meth:`parse` — so a table configured under
    one (e.g. for a raw file named ``sources.csv``) would otherwise be
    silently dropped from the event plan, losing every event it defines. An
    event-table block is recognized by its reserved sub-keys (``_defaults``
    / ``_table``) or by any event-shaped entry (a mapping with a ``code``
    key); neither shape occurs in legitimate reserved blocks, whose bucket
    values are lists and whose settings are scalars.

    Examples:
        >>> MessyConfig.parse({
        ...     "sources": {"_defaults": {"subject_id": "$sid"}, "e": {"code": "X", "time": None}},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: Top-level key 'sources' is reserved for download/ETL configuration, but its
        content looks like an event-table block (found '_defaults'; event-shaped entries ['e']).
        Tables cannot be named 'etl' or 'sources' — rename the source file, or nest it under a
        directory so the table prefix differs (e.g. 'raw/sources').

        A legitimate ``sources:`` block (buckets are lists, settings scalars) is untouched:

        >>> MessyConfig.parse({"sources": {"dataset": [], "dataset_version": "1.0"}}).sources_version
        '1.0'
    """
    if not isinstance(block, Mapping):
        return
    markers = sorted({"_defaults", "_table"} & set(block))
    event_like = sorted(k for k, v in block.items() if isinstance(v, Mapping) and "code" in v)
    if not markers and not event_like:
        return
    found = "; ".join(
        part
        for part in (
            ", ".join(f"'{m}'" for m in markers),
            f"event-shaped entries {event_like}" if event_like else "",
        )
        if part
    )
    names = " or ".join(f"'{k}'" for k in sorted(cls._RESERVED_TOP_LEVEL_KEYS))
    raise ValueError(
        f"Top-level key '{key}' is reserved for download/ETL configuration, but its content "
        f"looks like an event-table block (found {found}). Tables cannot be named {names} — "
        f"rename the source file, or nest it under a directory so the table prefix differs "
        f"(e.g. 'raw/{key}')."
    )

_sources_dataset_version(doc) staticmethod

Read + shape-validate the reserved sources.dataset_version node of a spec document.

Scalar string (one version for every bucket) or {bucket: version} mapping (demo/full releases genuinely differ). Being an ordinary document node it is interpolatable into source URLs (${sources.dataset_version}); only this node is resolved here. MessyConfig.save’s sources redaction strips it from output-tree copies — fine, since the stamped value lands durably in metadata/dataset.json.

Examples:

>>> doc = OmegaConf.create({"sources": {"dataset_version": "3.1"}})
>>> MessyConfig._sources_dataset_version(doc)
'3.1'
>>> MessyConfig._sources_dataset_version(OmegaConf.create({"patients": {}})) is None
True
>>> MessyConfig._sources_dataset_version(OmegaConf.create({"sources": {"dataset_version": 3.1}}))
Traceback (most recent call last):
    ...
ValueError: sources.dataset_version must be a version string (quote it in YAML:
dataset_version: "3.1") or a {bucket: version string} mapping, got float (3.1).

Source code in MEDS_extract/config.py

@staticmethod
def _sources_dataset_version(doc: DictConfig | Mapping | Any) -> str | dict[str, str] | None:
    """Read + shape-validate the reserved ``sources.dataset_version`` node of a spec document.

    Scalar string (one version for every bucket) or ``{bucket: version}`` mapping
    (demo/full releases genuinely differ). Being an ordinary document node it is
    interpolatable into source URLs (``${sources.dataset_version}``); only this node
    is resolved here. ``MessyConfig.save``'s sources redaction strips it from
    output-tree copies — fine, since the stamped value lands durably in
    ``metadata/dataset.json``.

    Examples:
        >>> doc = OmegaConf.create({"sources": {"dataset_version": "3.1"}})
        >>> MessyConfig._sources_dataset_version(doc)
        '3.1'
        >>> MessyConfig._sources_dataset_version(OmegaConf.create({"patients": {}})) is None
        True
        >>> MessyConfig._sources_dataset_version(OmegaConf.create({"sources": {"dataset_version": 3.1}}))
        Traceback (most recent call last):
            ...
        ValueError: sources.dataset_version must be a version string (quote it in YAML:
        dataset_version: "3.1") or a {bucket: version string} mapping, got float (3.1).
    """
    sources_node = doc.get("sources") if isinstance(doc, DictConfig | Mapping) else None
    if sources_node is None or "dataset_version" not in sources_node:
        return None
    # Scalar access through DictConfig resolves interpolations directly; a mapping
    # comes back as a DictConfig and needs an explicit resolving conversion.
    value = sources_node["dataset_version"]
    if OmegaConf.is_config(value):
        value = OmegaConf.to_container(value, resolve=True)
    ok = nonempty_str_err(value) is None or (
        isinstance(value, dict)
        and value
        and all(nonempty_str_err(b) is None and nonempty_str_err(v) is None for b, v in value.items())
    )
    if not ok:
        raise ValueError(
            f"sources.dataset_version must be a version string (quote it in YAML: "
            f'dataset_version: "{value}") or a {{bucket: version string}} mapping, got '
            f"{type(value).__name__} ({value!r})."
        )
    return value

dataset_version_for(key='dataset', override=None)

The etl_metadata.dataset_version stamp for one run.

{raw version}:{providing distribution's version} for registry-resolved specs; the raw version alone otherwise; override (the CLI’s dataset_version=) always wins.

Examples:

>>> MessyConfig(sources_version="3.1", dist_version="1.2.3").dataset_version_for()
'3.1:1.2.3'
>>> MessyConfig(sources_version="3.1").dataset_version_for(override="custom")
'custom'

Source code in MEDS_extract/config.py

def dataset_version_for(self, key: str = "dataset", override: str | None = None) -> str:
    """The ``etl_metadata.dataset_version`` stamp for one run.

    ``{raw version}:{providing distribution's version}`` for registry-resolved
    specs; the raw version alone otherwise; ``override`` (the CLI's
    ``dataset_version=``) always wins.

    Examples:
        >>> MessyConfig(sources_version="3.1", dist_version="1.2.3").dataset_version_for()
        '3.1:1.2.3'
        >>> MessyConfig(sources_version="3.1").dataset_version_for(override="custom")
        'custom'
    """
    if override:
        return override
    raw = self.raw_version_for(key)
    return f"{raw}:{self.dist_version}" if self.dist_version else raw

events_by_metadata_prefix()

Invert the events → metadata mapping.

Each event’s _metadata block maps metadata-file prefixes to per-prefix metadata config dicts. This returns the reverse: each metadata prefix gets the list of {code, _metadata, source_block} entries that reference it. The code value is always the original raw dftly expression string — a metadata-carrying event is guaranteed to retain it (enforced at :class:EventConfig construction), and it is what downstream code_template columns stamp verbatim. The source_block value is the {input_prefix}/{event_name} tag that :meth:EventConfig.extract stamps on every output row — extract_code_metadata uses it to scope metadata joins to the event that declared the _metadata block.

Used by extract_code_metadata.

Examples:

>>> cfg = MessyConfig.parse({
...     "_defaults": {"subject_id": "$MRN"},
...     "icu/procedureevents": {
...         "_defaults": {"subject_id": "$subject_id"},
...         "start": {
...             "code": 'f"PROC//START//{$itemid}"',
...             "_metadata": {
...                 "proc_datetimeevents": {
...                     "itemid": "$itemid",
...                     "desc": "coalesce($omop_concept_name, $label)",
...                 },
...             },
...         },
...     },
... })
>>> grouped = cfg.events_by_metadata_prefix()
>>> sorted(grouped.keys())
['proc_datetimeevents']
>>> entry = grouped["proc_datetimeevents"][0]
>>> entry["code"]
'f"PROC//START//{$itemid}"'
>>> entry["source_block"]
'icu/procedureevents/start'
>>> MessyConfig.parse({"t": {"e": {"code": "X", "time": None}}}).events_by_metadata_prefix()
{}

Source code in MEDS_extract/config.py

def events_by_metadata_prefix(self) -> dict[str, list[dict]]:
    """Invert the events → metadata mapping.

    Each event's ``_metadata`` block maps metadata-file prefixes to
    per-prefix metadata config dicts. This returns the reverse: each
    metadata prefix gets the list of ``{code, _metadata, source_block}``
    entries that reference it. The ``code`` value is always the original
    raw dftly expression string — a metadata-carrying event is guaranteed
    to retain it (enforced at :class:`EventConfig` construction), and it
    is what downstream ``code_template`` columns stamp verbatim. The
    ``source_block`` value is the
    ``{input_prefix}/{event_name}`` tag that :meth:`EventConfig.extract`
    stamps on every output row — ``extract_code_metadata`` uses it to
    scope metadata joins to the event that declared the ``_metadata``
    block.

    Used by ``extract_code_metadata``.

    Examples:
        >>> cfg = MessyConfig.parse({
        ...     "_defaults": {"subject_id": "$MRN"},
        ...     "icu/procedureevents": {
        ...         "_defaults": {"subject_id": "$subject_id"},
        ...         "start": {
        ...             "code": 'f"PROC//START//{$itemid}"',
        ...             "_metadata": {
        ...                 "proc_datetimeevents": {
        ...                     "itemid": "$itemid",
        ...                     "desc": "coalesce($omop_concept_name, $label)",
        ...                 },
        ...             },
        ...         },
        ...     },
        ... })
        >>> grouped = cfg.events_by_metadata_prefix()
        >>> sorted(grouped.keys())
        ['proc_datetimeevents']
        >>> entry = grouped["proc_datetimeevents"][0]
        >>> entry["code"]
        'f"PROC//START//{$itemid}"'
        >>> entry["source_block"]
        'icu/procedureevents/start'
        >>> MessyConfig.parse({"t": {"e": {"code": "X", "time": None}}}).events_by_metadata_prefix()
        {}
    """
    out: dict[str, list[dict]] = {}
    for table in self.event_tables:
        for event in table.events:
            source_block = f"{table.input_prefix}/{event.name}"
            for metadata_prefix, metadata_cfg in event.metadata.items():
                out.setdefault(metadata_prefix, []).append(
                    {"code": event.raw_code, "_metadata": metadata_cfg, SOURCE_BLOCK_COL: source_block}
                )
    return out

load(spec) classmethod

THE loading entry point: resolve a spec reference, read, validate, parse.

Every consumer — the run CLI, the download CLI, and all eight stages (via MESSY_config_fp) — funnels through this one call. spec resolves down the same ladder MEDS_transform-pipeline has for pipeline configs, extended one rung up with the registry:

1. Registered name — exact match against the MEDS_extract.pipelines entry-point group. The registration points directly at the bundled file ("<package.module>:<filename.yaml>"); its raw .value string is parsed here — never load()-ed, so registration cannot execute dataset-package code. The registered name and the providing distribution’s version land on :attr:registered_name / :attr:dist_version (feeding :attr:dataset_name and :meth:dataset_version_for), and :attr:spec_ref becomes the equivalent portable pkg:// reference.
2. pkg:// — resolved via the shared :func:resolve_config_path.
3. Explicit filesystem path — an absolute path, a ~ path, or an explicit relative path (./x.yaml / ../x.yaml). A bare name that matches no registration is an error naming both remedies (the registered names, and the ./ spelling) — it is NOT tried as a relative path, so a typo’d dataset name can never silently resolve to a stray local file, and an explicit path never collides with an entry-point name.

Examples:

>>> with yaml_disk('''
... spec.yaml: |
...   etl: {dataset_name: Example, raw_dataset_version: "0.1"}
...   patients:
...     dob: {code: BIRTH, time: null}
... ''') as d:
...     messy = MessyConfig.load(Path(d) / "spec.yaml")
...     (messy.dataset_name, messy.raw_version_for(), messy.spec_ref == str(messy.source_fp))
('Example', '0.1', True)

The registry rung, demonstrated against a synthetic installed package + registrations (the fake_pipeline_registry test fixture): the registered name supplies :attr:dataset_name, the distribution supplies the version-stamp suffix, and :attr:spec_ref is the portable pkg:// form:

>>> getfixture("fake_pipeline_registry")
>>> messy = MessyConfig.load("Fake-DS")
>>> (messy.dataset_name, messy.spec_ref, messy.dataset_version_for())
('Fake-DS', 'pkg://fake_ds_pkg.event_configs.yaml', '0.9:1.2.3')

Malformed registrations fail with targeted messages — a bare module reference (no :filename) and a registration naming a resource the module doesn’t bundle:

>>> MessyConfig.load("Bare-Mod")
Traceback (most recent call last):
    ...
ValueError: Entry point 'Bare-Mod' in group 'MEDS_extract.pipelines' has value
'fake_ds_pkg', which does not name the bundled MESSY file. Register it as
'<package.module>:<filename.yaml>', ...
>>> MessyConfig.load("Missing-Res")
Traceback (most recent call last):
    ...
ValueError: Entry point 'Missing-Res' points at 'fake_ds_pkg:nope.yaml', but module
'fake_ds_pkg' has no resource named 'nope.yaml' ...

A bare name matching no registration fails with both remedies — the registered names, and the explicit-path spelling (bare relative paths are deliberately not tried, so file.yaml errors while ./file.yaml loads):

>>> MessyConfig.load("Not-A-Registered-Name")
Traceback (most recent call last):
    ...
FileNotFoundError: spec='Not-A-Registered-Name' is not a registered pipeline name or a
pkg:// reference. Registered pipelines: Bare-Mod, Fake-DS, Missing-Res. To load a MESSY
file by path, give an absolute path or an explicit relative path
(./Not-A-Registered-Name).

An explicit path that does not exist says so directly:

>>> MessyConfig.load("./no-such-file.yaml")
Traceback (most recent call last):
    ...
FileNotFoundError: spec='./no-such-file.yaml' resolved to ...no-such-file.yaml, which
does not exist.

Source code in MEDS_extract/config.py

@classmethod
def load(cls, spec: str | Path) -> MessyConfig:
    """THE loading entry point: resolve a spec reference, read, validate, parse.

    Every consumer — the run CLI, the download CLI, and all eight stages (via
    ``MESSY_config_fp``) — funnels through this one call. ``spec``
    resolves down the same ladder ``MEDS_transform-pipeline`` has for pipeline
    configs, extended one rung up with the registry:

    1. **Registered name** — exact match against the ``MEDS_extract.pipelines``
       entry-point group. The registration points directly at the bundled file
       (``"<package.module>:<filename.yaml>"``); its raw ``.value`` string is
       parsed here — never ``load()``-ed, so registration cannot execute
       dataset-package code. The registered name and the providing
       distribution's version land on :attr:`registered_name` /
       :attr:`dist_version` (feeding :attr:`dataset_name` and
       :meth:`dataset_version_for`), and :attr:`spec_ref` becomes the
       equivalent portable ``pkg://`` reference.
    2. **pkg://** — resolved via the shared :func:`resolve_config_path`.
    3. **Explicit filesystem path** — an absolute path, a ``~`` path, or an
       explicit relative path (``./x.yaml`` / ``../x.yaml``). A bare name that
       matches no registration is an error naming both remedies (the registered
       names, and the ``./`` spelling) — it is NOT tried as a relative path, so
       a typo'd dataset name can never silently resolve to a stray local file,
       and an explicit path never collides with an entry-point name.

    Examples:
        >>> with yaml_disk('''
        ... spec.yaml: |
        ...   etl: {dataset_name: Example, raw_dataset_version: "0.1"}
        ...   patients:
        ...     dob: {code: BIRTH, time: null}
        ... ''') as d:
        ...     messy = MessyConfig.load(Path(d) / "spec.yaml")
        ...     (messy.dataset_name, messy.raw_version_for(), messy.spec_ref == str(messy.source_fp))
        ('Example', '0.1', True)

        The registry rung, demonstrated against a synthetic installed package +
        registrations (the ``fake_pipeline_registry`` test fixture): the
        registered name supplies :attr:`dataset_name`, the distribution supplies
        the version-stamp suffix, and :attr:`spec_ref` is the portable ``pkg://``
        form:

        >>> getfixture("fake_pipeline_registry")
        >>> messy = MessyConfig.load("Fake-DS")
        >>> (messy.dataset_name, messy.spec_ref, messy.dataset_version_for())
        ('Fake-DS', 'pkg://fake_ds_pkg.event_configs.yaml', '0.9:1.2.3')

        Malformed registrations fail with targeted messages — a bare module
        reference (no ``:filename``) and a registration naming a resource the
        module doesn't bundle:

        >>> MessyConfig.load("Bare-Mod")
        Traceback (most recent call last):
            ...
        ValueError: Entry point 'Bare-Mod' in group 'MEDS_extract.pipelines' has value
        'fake_ds_pkg', which does not name the bundled MESSY file. Register it as
        '<package.module>:<filename.yaml>', ...
        >>> MessyConfig.load("Missing-Res")
        Traceback (most recent call last):
            ...
        ValueError: Entry point 'Missing-Res' points at 'fake_ds_pkg:nope.yaml', but module
        'fake_ds_pkg' has no resource named 'nope.yaml' ...

        A bare name matching no registration fails with both remedies — the
        registered names, and the explicit-path spelling (bare relative paths are
        deliberately not tried, so ``file.yaml`` errors while ``./file.yaml``
        loads):

        >>> MessyConfig.load("Not-A-Registered-Name")
        Traceback (most recent call last):
            ...
        FileNotFoundError: spec='Not-A-Registered-Name' is not a registered pipeline name or a
        pkg:// reference. Registered pipelines: Bare-Mod, Fake-DS, Missing-Res. To load a MESSY
        file by path, give an absolute path or an explicit relative path
        (./Not-A-Registered-Name).

        An explicit path that does not exist says so directly:

        >>> MessyConfig.load("./no-such-file.yaml")  # doctest: +ELLIPSIS
        Traceback (most recent call last):
            ...
        FileNotFoundError: spec='./no-such-file.yaml' resolved to ...no-such-file.yaml, which
        does not exist.
    """
    spec = str(spec)
    by_name = {ep.name: ep for ep in entry_points(group=cls.PIPELINES_ENTRY_POINT_GROUP)}
    registered_name = dist_version = None
    if spec in by_name:
        ep = by_name[spec]
        module, _, filename = ep.value.partition(":")
        if not module or not filename:
            raise ValueError(
                f"Entry point {spec!r} in group {cls.PIPELINES_ENTRY_POINT_GROUP!r} has value "
                f"{ep.value!r}, which does not name the bundled MESSY file. Register it as "
                f"'<package.module>:<filename.yaml>', e.g. "
                f"'MIMIC_IV_MEDS.configs:event_configs.yaml'."
            )
        spec_ref = f"{PKG_PFX}{module}.{filename.replace('/', '.')}"
        fp = Path(str(files(module) / filename))
        if not fp.is_file():
            raise ValueError(
                f"Entry point {spec!r} points at {ep.value!r}, but module {module!r} has no "
                f"resource named {filename!r} (resolved to {fp})."
            )
        registered_name = ep.name
        dist_version = ep.dist.version if ep.dist is not None else None
    elif spec.startswith(PKG_PFX):
        spec_ref = spec
        fp = resolve_config_path(spec)
        if not fp.is_file():
            raise FileNotFoundError(f"spec={spec!r} resolved to {fp}, which does not exist.")
    elif spec.startswith(("./", "../", "~")) or Path(spec).is_absolute():
        fp = Path(spec).expanduser().resolve()
        if not fp.is_file():
            raise FileNotFoundError(f"spec={spec!r} resolved to {fp}, which does not exist.")
        spec_ref = str(fp)
    else:
        # A bare name is ONLY a registry lookup: never fall back to treating it as
        # a relative path, so a typo'd dataset name cannot silently resolve to a
        # stray local file (and an explicit path never collides with a name).
        registered = ", ".join(sorted(by_name)) or "(none)"
        raise FileNotFoundError(
            f"spec={spec!r} is not a registered pipeline name or a pkg:// reference. "
            f"Registered pipelines: {registered}. To load a MESSY file by path, give an "
            f"absolute path or an explicit relative path (./{spec})."
        )

    logger.info(f"Reading MESSY config from {fp}")
    raw = OmegaConf.load(fp)
    # Log with credential-bearing reserved keys stripped: a combined-MESSY
    # ``sources:`` block can carry credentials (literal API keys / passwords),
    # which must not land in every stage's log output. The ``etl:`` block is
    # credential-free and stays in the log — it's useful provenance.
    loggable = OmegaConf.create(raw)
    for key in cls._CREDENTIALED_TOP_LEVEL_KEYS:
        if key in loggable:
            del loggable[key]
    logger.info(f"MESSY config:\n{OmegaConf.to_yaml(loggable)}")
    parsed = cls.parse(raw)
    # Attach the spec context (frozen dataclass => object.__setattr__, the same
    # idiom the pre-existing source_fp attachment used).
    for attr, value in [
        ("source_fp", fp),
        ("spec_ref", spec_ref),
        ("registered_name", registered_name),
        ("dist_version", dist_version),
    ]:
        object.__setattr__(parsed, attr, value)
    return parsed

needed_source_columns()

Map each source prefix to the sorted list of columns that must be read.

Aggregates each table’s own :meth:TableConfig.source_columns plus the columns that any table pulls in from a join target. The returned dict is the input to convert_to_parquet — it tells that stage which columns to project when normalizing raw inputs.

Examples:

>>> cfg = MessyConfig.parse({
...     "_defaults": {"subject_id": "$subject_id_global"},
...     "hosp/patients": {
...         "eye_color": {"code": "EYE_COLOR", "time": None},
...         "height": {"code": "HEIGHT", "time": None, "numeric_value": "$height"},
...     },
...     "icu/chartevents": {
...         "_defaults": {"subject_id": "$subject_id_icu"},
...         "heart_rate": {
...             "code": "HEART_RATE", "time": "$charttime", "numeric_value": "$HR"
...         },
...     },
... })
>>> cfg.needed_source_columns()
{'hosp/patients': ['height', 'subject_id_global'],
 'icu/chartevents': ['HR', 'charttime', 'subject_id_icu']}

Derived columns and joined columns are excluded from the source file’s needed-column list (they come from elsewhere); the join target gets its own entry:

>>> cfg = MessyConfig.parse({
...     "labs": {
...         "_defaults": {"subject_id": "$patient_id"},
...         "_table": {"join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}}},
...         "lab": {"code": "$test", "time": "$dischtime"},
...     },
... })
>>> cfg.needed_source_columns()
{'labs': ['patient_id', 'stay_id', 'test'], 'stays': ['dischtime', 'stay_id']}

Aggregated joins plan the same way: the aggregation’s source column is needed on the right-side table even though the left-side table never reads it directly — deathtime below is the input to min() on the admissions side:

>>> cfg = MessyConfig.parse({
...     "hosp/patients": {
...         "_table": {
...             "join": {
...                 "hosp/admissions": {"key": "subject_id", "cols": {"deathtime": "min"}}
...             }
...         },
...         "death": {"code": "MEDS_DEATH", "time": "$deathtime"},
...     },
... })
>>> cfg.needed_source_columns()
{'hosp/patients': ['subject_id'], 'hosp/admissions': ['deathtime', 'subject_id']}

Transform outputs are computed at read time, not read from disk, so they are excluded from the plan while their input columns are included. Format-annotated time strings contribute their source column, and _metadata blocks contribute nothing:

>>> cfg = MessyConfig.parse({
...     "hosp/patients": {
...         "_table": {"cols": {"year_of_birth": "$anchor_year - $anchor_age"}},
...         "dob": {"code": "MEDS_BIRTH", "time": "$year_of_birth::year"},
...         "admit": {
...             "code": 'f"ADMIT//{$admtype}"',
...             "time": '$admittime::"%Y-%m-%d %H:%M:%S"',
...             "_metadata": {
...                 "admissions_meta": {"admtype": "$admtype", "description": "$adm_desc"}
...             },
...         },
...     },
... })
>>> cfg.needed_source_columns()
{'hosp/patients': ['admittime', 'admtype', 'anchor_age', 'anchor_year', 'subject_id']}

Source code in MEDS_extract/config.py

def needed_source_columns(self) -> dict[str, list[str]]:
    """Map each source prefix to the sorted list of columns that must be read.

    Aggregates each table's own :meth:`TableConfig.source_columns` plus the
    columns that any table pulls in from a join target. The returned dict is
    the input to ``convert_to_parquet`` — it tells that stage which columns
    to project when normalizing raw inputs.

    Examples:
        >>> cfg = MessyConfig.parse({
        ...     "_defaults": {"subject_id": "$subject_id_global"},
        ...     "hosp/patients": {
        ...         "eye_color": {"code": "EYE_COLOR", "time": None},
        ...         "height": {"code": "HEIGHT", "time": None, "numeric_value": "$height"},
        ...     },
        ...     "icu/chartevents": {
        ...         "_defaults": {"subject_id": "$subject_id_icu"},
        ...         "heart_rate": {
        ...             "code": "HEART_RATE", "time": "$charttime", "numeric_value": "$HR"
        ...         },
        ...     },
        ... })
        >>> cfg.needed_source_columns()
        {'hosp/patients': ['height', 'subject_id_global'],
         'icu/chartevents': ['HR', 'charttime', 'subject_id_icu']}

        Derived columns and joined columns are excluded from the source file's
        needed-column list (they come from elsewhere); the join target gets
        its own entry:

        >>> cfg = MessyConfig.parse({
        ...     "labs": {
        ...         "_defaults": {"subject_id": "$patient_id"},
        ...         "_table": {"join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}}},
        ...         "lab": {"code": "$test", "time": "$dischtime"},
        ...     },
        ... })
        >>> cfg.needed_source_columns()
        {'labs': ['patient_id', 'stay_id', 'test'], 'stays': ['dischtime', 'stay_id']}

        Aggregated joins plan the same way: the aggregation's
        source column is needed on the *right*-side table even though the
        left-side table never reads it directly — ``deathtime`` below is the
        input to ``min()`` on the admissions side:

        >>> cfg = MessyConfig.parse({
        ...     "hosp/patients": {
        ...         "_table": {
        ...             "join": {
        ...                 "hosp/admissions": {"key": "subject_id", "cols": {"deathtime": "min"}}
        ...             }
        ...         },
        ...         "death": {"code": "MEDS_DEATH", "time": "$deathtime"},
        ...     },
        ... })
        >>> cfg.needed_source_columns()
        {'hosp/patients': ['subject_id'], 'hosp/admissions': ['deathtime', 'subject_id']}

        Transform *outputs* are computed at read time, not read from disk, so they are
        excluded from the plan while their input columns are included.
        Format-annotated time strings contribute their source column, and ``_metadata``
        blocks contribute nothing:

        >>> cfg = MessyConfig.parse({
        ...     "hosp/patients": {
        ...         "_table": {"cols": {"year_of_birth": "$anchor_year - $anchor_age"}},
        ...         "dob": {"code": "MEDS_BIRTH", "time": "$year_of_birth::year"},
        ...         "admit": {
        ...             "code": 'f"ADMIT//{$admtype}"',
        ...             "time": '$admittime::"%Y-%m-%d %H:%M:%S"',
        ...             "_metadata": {
        ...                 "admissions_meta": {"admtype": "$admtype", "description": "$adm_desc"}
        ...             },
        ...         },
        ...     },
        ... })
        >>> cfg.needed_source_columns()
        {'hosp/patients': ['admittime', 'admtype', 'anchor_age', 'anchor_year', 'subject_id']}
    """
    out: dict[str, set[str]] = {}
    for table in self.event_tables:
        out.setdefault(table.input_prefix, set()).update(table.source_columns())
        if table.join is not None:
            jt = out.setdefault(table.join.input_prefix, set())
            jt.update(table.join.right_on)
            jt.update(table.join.cols)
    return {k: sorted(v) for k, v in out.items()}

parse(raw) classmethod

Parse a raw MESSY mapping into a :class:MessyConfig.

Every present section is validated here; absent sections parse to their neutral field values (tables=(), all-defaults :attr:etl, sources_version=None) and mis-ACCESS of an absent section raises later, at the accessor. The reserved sections are captured before the reserved keys are stripped, and stripping happens before interpolation resolution, so a download-only ${oc.env:...} inside sources: never requires its env var to be set just to load the event-conversion side:

cfg = MessyConfig.parse(OmegaConf.create({ … “sources”: {“dataset”: [{“type”: “fsspec”, “root”: “\({oc.env:UNSET_DOWNLOAD_ROOT}"}]}, ... "_defaults": {"subject_id": "\)patient_id”}, … “patients”: {“dob”: {“code”: “DOB”, “time”: “$dob”}}, … })) cfg.table_prefixes [‘patients’]

The reserved sections land on their own fields, parsed in full-document context — config mistakes in them fail at load time in every consumer:

cfg = MessyConfig.parse({ … “sources”: {“dataset_version”: “0.1”}, … “etl”: {“dataset_name”: “Example”}, … “patients”: {“dob”: {“code”: “DOB”, “time”: “\(dob"}}, ... }) (cfg.etl.dataset_name, cfg.sources_version) ('Example', '0.1') MessyConfig.parse({ ... "etl": {"pipeline": ["convert_to_parquet"]}, ... "patients": {"dob": {"code": "DOB", "time": "\)dob”}}, … }) Traceback (most recent call last): … ValueError: etl: block contains unknown key(s) [‘pipeline’]. …

A sources-only document is valid (it can drive meds-extract-download); the event-conversion side errors at ACCESS, not load:

cfg = MessyConfig.parse({“sources”: {“dataset”: []}}) cfg.event_tables Traceback (most recent call last): … ValueError: This MESSY spec declares no event tables …

Event-section mistakes surface at the first :attr:event_tables access — i.e. immediately after load in every event-conversion consumer, and never for the download CLI (whose spec may legitimately carry unresolvable event-side interpolations). _metadata mistakes, for example, name the event, the prefix, and the components the event offers:

cfg = MessyConfig.parse({ … “chartevents”: { … “chart”: { … “code”: ‘f”CHART//{\(itemid}"', ... "time": None, ... "_metadata": {"d_items": {"description": "\)label”}}, … }, … }, … }) cfg.event_tables Traceback (most recent call last): … ValueError: _metadata block (event ‘chart’, metadata prefix ‘d_items’) produces no join-key columns: … Component columns available on this event: [‘itemid’] …

Source code in MEDS_extract/config.py

@classmethod
def parse(cls, raw: Mapping[str, Any] | DictConfig) -> MessyConfig:
    """Parse a raw MESSY mapping into a :class:`MessyConfig`.

    Every present section is validated here; absent sections parse to their
    neutral field values (``tables=()``, all-defaults :attr:`etl`,
    ``sources_version=None``) and mis-ACCESS of an absent section raises later,
    at the accessor. The reserved sections are captured before the reserved
    keys are stripped, and stripping happens *before* interpolation resolution,
    so a download-only ``${oc.env:...}`` inside ``sources:`` never requires its
    env var to be set just to load the event-conversion side:

    >>> cfg = MessyConfig.parse(OmegaConf.create({
    ...     "sources": {"dataset": [{"type": "fsspec", "root": "${oc.env:UNSET_DOWNLOAD_ROOT}"}]},
    ...     "_defaults": {"subject_id": "$patient_id"},
    ...     "patients": {"dob": {"code": "DOB", "time": "$dob"}},
    ... }))
    >>> cfg.table_prefixes
    ['patients']

    The reserved sections land on their own fields, parsed in full-document
    context — config mistakes in them fail at load time in every consumer:

    >>> cfg = MessyConfig.parse({
    ...     "sources": {"dataset_version": "0.1"},
    ...     "etl": {"dataset_name": "Example"},
    ...     "patients": {"dob": {"code": "DOB", "time": "$dob"}},
    ... })
    >>> (cfg.etl.dataset_name, cfg.sources_version)
    ('Example', '0.1')
    >>> MessyConfig.parse({
    ...     "etl": {"pipeline": ["convert_to_parquet"]},
    ...     "patients": {"dob": {"code": "DOB", "time": "$dob"}},
    ... })
    Traceback (most recent call last):
        ...
    ValueError: etl: block contains unknown key(s) ['pipeline']. ...

    A sources-only document is valid (it can drive ``meds-extract-download``);
    the event-conversion side errors at ACCESS, not load:

    >>> cfg = MessyConfig.parse({"sources": {"dataset": []}})
    >>> cfg.event_tables
    Traceback (most recent call last):
        ...
    ValueError: This MESSY spec declares no event tables ...

    Event-section mistakes surface at the first :attr:`event_tables` access —
    i.e. immediately after load in every event-conversion consumer, and never
    for the download CLI (whose spec may legitimately carry unresolvable
    event-side interpolations). ``_metadata`` mistakes, for example, name the
    event, the prefix, and the components the event offers:

    >>> cfg = MessyConfig.parse({
    ...     "chartevents": {
    ...         "chart": {
    ...             "code": 'f"CHART//{$itemid}"',
    ...             "time": None,
    ...             "_metadata": {"d_items": {"description": "$label"}},
    ...         },
    ...     },
    ... })
    >>> cfg.event_tables
    Traceback (most recent call last):
        ...
    ValueError: _metadata block (event 'chart', metadata prefix 'd_items') produces no join-key
    columns: ... Component columns available on this event: ['itemid'] ...
    """
    # Keep the pristine, UNRESOLVED document for accessor-time resolution
    # (``selected_sources`` / ``event_tables``); everything below works on copies.
    raw_doc = OmegaConf.create(raw)
    for key in cls._RESERVED_TOP_LEVEL_KEYS:
        if key in raw_doc:
            cls._reject_table_shaped_reserved_block(
                key, OmegaConf.to_container(raw_doc[key], resolve=False)
            )
    sources_version = cls._sources_dataset_version(raw)
    etl_raw = OmegaConf.to_container(raw_doc.etl, resolve=False) if "etl" in raw_doc else None

    # The event-conversion remainder: reserved keys stripped, NOT resolved —
    # its interpolations resolve at ``event_tables`` access, so a download-only
    # consumer never needs event-side env vars and vice versa.
    stripped = OmegaConf.create(raw_doc)
    for key in cls._RESERVED_TOP_LEVEL_KEYS:
        if key in stripped:
            del stripped[key]

    return cls(
        etl=EtlConfig.parse(etl_raw),
        sources_version=sources_version,
        raw_doc=raw_doc,
        tables_raw=stripped,
    )

pipeline_config(*, input_dir, output_dir, key='dataset', dataset_version=None)

The full MEDS-transforms pipeline config for one run, as a plain dict.

Only meaningful on :meth:load-built instances (it needs the spec context). Every value is inlined as a resolved literal — no env-var indirection — so the written file is self-contained, diffable provenance and the only channel through which the computed identity reaches the MEDS_transform-pipeline subprocess. MESSY_config_fp carries :attr:spec_ref (the portable pkg:// form for registered/pkg specs), which every consumer resolves via this class’s ladder-aware :meth:load.

Examples:

>>> messy = MessyConfig(
...     etl=EtlConfig.parse({"n_subjects_per_shard": 1000}),
...     sources_version="3.1",
...     dist_version="1.0.0",
...     registered_name="Example",
...     spec_ref="pkg://example_pkg.messy.yaml",
... )
>>> cfg = messy.pipeline_config(
...     input_dir=Path("/data/raw_input"), output_dir=Path("/data/MEDS_cohort")
... )
>>> print(OmegaConf.to_yaml(OmegaConf.create(cfg)).strip())
etl_metadata:
  dataset_name: Example
  dataset_version: 3.1:1.0.0
MESSY_config_fp: pkg://example_pkg.messy.yaml
input_dir: /data/raw_input
output_dir: /data/MEDS_cohort
shards_map_fp: /data/MEDS_cohort/metadata/.shards.json
stages:
- convert_to_parquet
- split_and_shard_subjects:
    n_subjects_per_shard: 1000
- convert_to_subject_sharded
- convert_to_MEDS_events
- extract_code_metadata
- merge_to_MEDS_cohort
- finalize_MEDS_metadata
- finalize_MEDS_data

Source code in MEDS_extract/config.py

def pipeline_config(
    self,
    *,
    input_dir: Path,
    output_dir: Path,
    key: str = "dataset",
    dataset_version: str | None = None,
) -> dict[str, Any]:
    """The full MEDS-transforms pipeline config for one run, as a plain dict.

    Only meaningful on :meth:`load`-built instances (it needs the spec
    context). Every value is **inlined as a resolved literal** — no env-var
    indirection — so the written file is self-contained, diffable provenance
    and the only channel through which the computed identity reaches the
    ``MEDS_transform-pipeline`` subprocess. ``MESSY_config_fp``
    carries :attr:`spec_ref` (the portable ``pkg://`` form for registered/pkg
    specs), which every consumer resolves via this class's ladder-aware
    :meth:`load`.

    Examples:
        >>> messy = MessyConfig(
        ...     etl=EtlConfig.parse({"n_subjects_per_shard": 1000}),
        ...     sources_version="3.1",
        ...     dist_version="1.0.0",
        ...     registered_name="Example",
        ...     spec_ref="pkg://example_pkg.messy.yaml",
        ... )
        >>> cfg = messy.pipeline_config(
        ...     input_dir=Path("/data/raw_input"), output_dir=Path("/data/MEDS_cohort")
        ... )
        >>> print(OmegaConf.to_yaml(OmegaConf.create(cfg)).strip())
        etl_metadata:
          dataset_name: Example
          dataset_version: 3.1:1.0.0
        MESSY_config_fp: pkg://example_pkg.messy.yaml
        input_dir: /data/raw_input
        output_dir: /data/MEDS_cohort
        shards_map_fp: /data/MEDS_cohort/metadata/.shards.json
        stages:
        - convert_to_parquet
        - split_and_shard_subjects:
            n_subjects_per_shard: 1000
        - convert_to_subject_sharded
        - convert_to_MEDS_events
        - extract_code_metadata
        - merge_to_MEDS_cohort
        - finalize_MEDS_metadata
        - finalize_MEDS_data
    """
    return {
        "etl_metadata": {
            "dataset_name": self.dataset_name,
            "dataset_version": self.dataset_version_for(key, override=dataset_version),
        },
        "MESSY_config_fp": self.spec_ref,
        "input_dir": str(input_dir),
        "output_dir": str(output_dir),
        "shards_map_fp": f"{output_dir}/metadata/.shards.json",
        "stages": self.etl.stages_container(),
    }

raw_version_for(key='dataset')

The effective raw-data version for one run (selected bucket key).

sources.dataset_version (scalar, or the mapping’s entry for key) is authoritative when present; etl.raw_dataset_version is the fallback. When both resolve they must match — silent divergence would stamp a lie into dataset.json.

Examples:

>>> MessyConfig(sources_version={"dataset": "3.1", "demo": "2.2"}).raw_version_for("demo")
'2.2'
>>> MessyConfig(etl=EtlConfig(raw_dataset_version="0.1")).raw_version_for()
'0.1'
>>> MessyConfig(sources_version="3.1", etl=EtlConfig(raw_dataset_version="9.9")
...     ).raw_version_for()
Traceback (most recent call last):
    ...
ValueError: sources.dataset_version resolves to '3.1' for key='dataset' but
etl.raw_dataset_version says '9.9'. These must match — keep one source of truth
(prefer sources.dataset_version and drop the etl: fallback).
>>> MessyConfig().raw_version_for()
Traceback (most recent call last):
    ...
ValueError: No raw dataset version declared: add `dataset_version` to the sources:
block (scalar or per-bucket mapping), or `raw_dataset_version` to the etl: block.

Source code in MEDS_extract/config.py

def raw_version_for(self, key: str = "dataset") -> str:
    """The effective raw-data version for one run (selected bucket ``key``).

    ``sources.dataset_version`` (scalar, or the mapping's entry for ``key``) is
    authoritative when present; ``etl.raw_dataset_version`` is the fallback.
    When both resolve they must match — silent divergence would stamp a lie
    into ``dataset.json``.

    Examples:
        >>> MessyConfig(sources_version={"dataset": "3.1", "demo": "2.2"}).raw_version_for("demo")
        '2.2'
        >>> MessyConfig(etl=EtlConfig(raw_dataset_version="0.1")).raw_version_for()
        '0.1'
        >>> MessyConfig(sources_version="3.1", etl=EtlConfig(raw_dataset_version="9.9")
        ...     ).raw_version_for()
        Traceback (most recent call last):
            ...
        ValueError: sources.dataset_version resolves to '3.1' for key='dataset' but
        etl.raw_dataset_version says '9.9'. These must match — keep one source of truth
        (prefer sources.dataset_version and drop the etl: fallback).
        >>> MessyConfig().raw_version_for()
        Traceback (most recent call last):
            ...
        ValueError: No raw dataset version declared: add `dataset_version` to the sources:
        block (scalar or per-bucket mapping), or `raw_dataset_version` to the etl: block.
    """
    from_sources = (
        self.sources_version.get(key)
        if isinstance(self.sources_version, Mapping)
        else self.sources_version
    )
    if from_sources and self.etl.raw_dataset_version and from_sources != self.etl.raw_dataset_version:
        raise ValueError(
            f"sources.dataset_version resolves to {from_sources!r} for key={key!r} but "
            f"etl.raw_dataset_version says {self.etl.raw_dataset_version!r}. These must "
            f"match — keep one source of truth (prefer sources.dataset_version and drop "
            f"the etl: fallback)."
        )
    effective = from_sources or self.etl.raw_dataset_version
    if effective is None:
        raise ValueError(
            "No raw dataset version declared: add `dataset_version` to the sources: block "
            "(scalar or per-bucket mapping), or `raw_dataset_version` to the etl: block."
        )
    return effective

save(fp)

Copy the original MESSY config file to fp, minus credentialed keys.

Only valid on instances produced by :meth:load (which remembers the source path). Instances built via :meth:parse directly don’t have a source file to copy and will raise. Uses read_bytes / write_bytes so UPath-backed cloud destinations work as well as local paths.

When the source file carries credential-bearing reserved blocks (sources:), the copy is re-serialized with those blocks stripped — a combined-MESSY sources: block can carry credentials, and this copy lands inside the (often shared) pipeline output tree. The credential-free etl: block is NOT stripped: the stage list and dataset name/version are useful provenance in the output copy. Comment formatting is preserved only for files with no credentialed blocks, where a verbatim byte-copy suffices.

Examples:

>>> yaml = '''
... sources:
...   dataset:
...     - type: http
...       headers: {X-Dataverse-key: super-secret-token}
...       urls: [https://example.com/x.csv]
... etl:
...   dataset_name: Example
...   raw_dataset_version: "0.1"
... patients:
...   dob: {code: BIRTH, time: null}
... '''
>>> cfg_fp = getfixture("tmp_path") / "cfg.yaml"
>>> _ = cfg_fp.write_text(yaml)
>>> out_fp = getfixture("tmp_path") / "copy.yaml"
>>> MessyConfig.load(cfg_fp).save(out_fp)
>>> print(out_fp.read_text().strip())
etl:
  dataset_name: Example
  raw_dataset_version: '0.1'
patients:
  dob:
    code: BIRTH
    time: null
>>> "super-secret-token" in out_fp.read_text()
False

Source code in MEDS_extract/config.py

def save(self, fp: Path | UPath | str) -> None:
    """Copy the original MESSY config file to ``fp``, minus credentialed keys.

    Only valid on instances produced by :meth:`load` (which remembers the
    source path). Instances built via :meth:`parse` directly don't have a
    source file to copy and will raise. Uses ``read_bytes`` / ``write_bytes``
    so UPath-backed cloud destinations work as well as local paths.

    When the source file carries credential-bearing reserved blocks
    (``sources:``), the copy is re-serialized with those blocks stripped — a
    combined-MESSY ``sources:`` block can carry credentials, and this copy lands
    inside the (often shared) pipeline output tree. The credential-free ``etl:``
    block is NOT stripped: the stage list and dataset name/version are useful
    provenance in the output copy. Comment formatting is preserved only for
    files with no credentialed blocks, where a verbatim byte-copy suffices.

    Examples:
        >>> yaml = '''
        ... sources:
        ...   dataset:
        ...     - type: http
        ...       headers: {X-Dataverse-key: super-secret-token}
        ...       urls: [https://example.com/x.csv]
        ... etl:
        ...   dataset_name: Example
        ...   raw_dataset_version: "0.1"
        ... patients:
        ...   dob: {code: BIRTH, time: null}
        ... '''
        >>> cfg_fp = getfixture("tmp_path") / "cfg.yaml"
        >>> _ = cfg_fp.write_text(yaml)
        >>> out_fp = getfixture("tmp_path") / "copy.yaml"
        >>> MessyConfig.load(cfg_fp).save(out_fp)
        >>> print(out_fp.read_text().strip())
        etl:
          dataset_name: Example
          raw_dataset_version: '0.1'
        patients:
          dob:
            code: BIRTH
            time: null
        >>> "super-secret-token" in out_fp.read_text()
        False
    """
    if self.source_fp is None:
        raise ValueError("MessyConfig.save requires a source file path (only available after .load()).")
    if not self.source_fp.exists():
        raise FileNotFoundError(
            f"MessyConfig source file no longer exists at {self.source_fp}; cannot copy to {fp}."
        )
    dest = Path(fp) if isinstance(fp, str) else fp
    raw = OmegaConf.load(self.source_fp)
    credentialed_present = [k for k in self._CREDENTIALED_TOP_LEVEL_KEYS if k in raw]
    if not credentialed_present:
        dest.write_bytes(self.source_fp.read_bytes())
        return
    for key in credentialed_present:
        del raw[key]
    # ``to_yaml`` does not resolve interpolations, so symbolic ``${oc.env:...}``
    # references in the event-conversion sections survive the round-trip.
    dest.write_bytes(OmegaConf.to_yaml(raw).encode("utf-8"))

selected_sources(key='dataset')

Construct the Source objects for one download run (bucket key + common).

Resolution is per-selected-bucket and happens here, at accessor time, against the raw unresolved document kept on the instance: only the key bucket and the always-appended common bucket are resolved, each while still ATTACHED to the document (so document-relative references like ${sources.dataset_version} keep resolving) — unselected buckets’ ${oc.env:...} credentials are never required.

Returns [] for a document with no sources: block — deliberately not an access-time error, because “nothing to download” is a legal state with defined semantics for this accessor’s one consumer (a download-free ETL warns and exits 0), unlike a table-less pipeline run. A key naming no bucket — including the reserved dataset_version metadata key — IS a config error (likely a typo): because common is always appended, a typo’d key would otherwise quietly fetch only the common bucket and “succeed”.

Examples:

>>> with yaml_disk('''
... mirror:
...   patients.csv: "patient_id\\n1\\n"
... ''') as d:
...     spec_fp = Path(d) / "spec.yaml"
...     src_yaml = f"sources:\n  dataset:\n    - type: fsspec\n      root: {d}/mirror\n"
...     _ = spec_fp.write_text(src_yaml)
...     [type(s).__name__ for s in MessyConfig.load(spec_fp).selected_sources()]
['FsspecSource']

>>> MessyConfig.parse({"patients": {"dob": {"code": "BIRTH", "time": None}}}
...     ).selected_sources()
[]

>>> MessyConfig.parse({"sources": {"dataset": []}}).selected_sources(key="dataste")
Traceback (most recent call last):
    ...
ValueError: key='dataste' does not name a sources bucket. Available buckets:
['dataset'].

Source code in MEDS_extract/config.py

def selected_sources(self, key: str = "dataset") -> list:
    """Construct the ``Source`` objects for one download run (bucket ``key`` + ``common``).

    Resolution is per-**selected**-bucket and happens here, at accessor time,
    against the raw unresolved document kept on the instance: only the ``key``
    bucket and the always-appended ``common`` bucket are resolved, each while
    still ATTACHED to the document (so document-relative references like
    ``${sources.dataset_version}`` keep resolving) — unselected buckets'
    ``${oc.env:...}`` credentials are never required.

    Returns ``[]`` for a document with no ``sources:`` block — deliberately not
    an access-time error, because "nothing to download" is a *legal state with
    defined semantics* for this accessor's one consumer (a download-free ETL
    warns and exits 0), unlike a table-less pipeline run. A ``key`` naming no
    bucket — including the reserved ``dataset_version`` metadata key — IS a
    config error (likely a typo): because ``common`` is always appended, a
    typo'd key would otherwise quietly fetch only the common bucket and
    "succeed".

    Examples:
        >>> with yaml_disk('''
        ... mirror:
        ...   patients.csv: "patient_id\\\\n1\\\\n"
        ... ''') as d:
        ...     spec_fp = Path(d) / "spec.yaml"
        ...     src_yaml = f"sources:\\n  dataset:\\n    - type: fsspec\\n      root: {d}/mirror\\n"
        ...     _ = spec_fp.write_text(src_yaml)
        ...     [type(s).__name__ for s in MessyConfig.load(spec_fp).selected_sources()]
        ['FsspecSource']

        >>> MessyConfig.parse({"patients": {"dob": {"code": "BIRTH", "time": None}}}
        ...     ).selected_sources()
        []

        >>> MessyConfig.parse({"sources": {"dataset": []}}).selected_sources(key="dataste")
        Traceback (most recent call last):
            ...
        ValueError: key='dataste' does not name a sources bucket. Available buckets:
        ['dataset'].
    """
    # Deferred: the download layer imports this module already (the reverse
    # import at module scope would be circular).
    from .download.spec import SOURCES_RESERVED_KEYS, sources_from_spec

    sources_node = self.raw_doc.get("sources") if self.raw_doc is not None else None
    # Bucket names come from the UNRESOLVED node — listing them must not require
    # any interpolation (in any bucket) to be resolvable.
    if sources_node and (key in SOURCES_RESERVED_KEYS or key not in sources_node):
        raise ValueError(
            f"key={key!r} does not name a sources bucket. "
            f"Available buckets: {sorted(set(sources_node) - SOURCES_RESERVED_KEYS)}."
        )
    sources_dict = {}
    if sources_node is not None:
        for bucket in dict.fromkeys((key, "common")):  # de-dupe when key="common"
            bucket_node = sources_node.get(bucket)
            if bucket_node is not None:
                sources_dict[bucket] = OmegaConf.to_container(bucket_node, resolve=True)
    return sources_from_spec({"sources": sources_dict}, key=key)

shuffled_tables()

Return tables in randomized order.

Used by stages that iterate tables to spread parallel worker load — without shuffling, every worker would contend on the same first table.

Source code in MEDS_extract/config.py

def shuffled_tables(self) -> list[TableConfig]:
    """Return tables in randomized order.

    Used by stages that iterate tables to spread parallel worker load — without shuffling, every worker
    would contend on the same first table.
    """
    tables = list(self.event_tables)
    random.Random().shuffle(tables)
    return tables

TableConfig dataclass

Fully-resolved config for one source table block.

Global _defaults are merged with file-level _defaults at parse time, so :attr:subject_id_node already reflects the final inherited value. _table sub-keys (cols, join) are lifted to top-level fields.

Examples:

>>> tc = TableConfig.parse("patients", {
...     "_defaults": {"subject_id": "$MRN"},
...     "_table": {"join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}}},
...     "dob": {"code": "BIRTH", "time": "$dob"},
... })
>>> tc.input_prefix
'patients'
>>> type(tc.subject_id_node).__name__
'Column'
>>> sorted(tc.subject_id_node.referenced_columns)
['MRN']
>>> tc.join.input_prefix
'stays'
>>> [e.name for e in tc.events]
['dob']

_defaults accepts only the keys it actually consumes — a stray key would otherwise be silently dropped. The sharpest case is time: an event with no time key is legally static, so a dropped default time would turn every event in the table static without a word:

>>> TableConfig.parse("vitals", {
...     "_defaults": {"subject_id": "$MRN", "time": '$charttime::"%Y-%m-%d"'},
...     "hr": {"code": "HR", "numeric_value": "$hr"},
... })
Traceback (most recent call last):
    ...
ValueError: Table 'vitals' has unknown keys under '_defaults': ['time']. Allowed keys:
'subject_id'.

A typo’d subject_id is caught the same way, instead of silently falling back to reading a literal subject_id source column:

>>> TableConfig.parse("vitals", {
...     "_defaults": {"subject_ids": "$MRN"},
...     "hr": {"code": "HR", "time": None},
... })
Traceback (most recent call last):
    ...
ValueError: Table 'vitals' has unknown keys under '_defaults': ['subject_ids']. Allowed
keys: 'subject_id'.

Underscore-prefixed keys other than _defaults/_table are almost certainly typos of those reserved names, and are rejected up front rather than falling through to event parsing with a misleading error:

>>> TableConfig.parse("vitals", {
...     "_default": {"subject_id": "$MRN"},
...     "hr": {"code": "HR", "time": None},
... })
Traceback (most recent call last):
    ...
ValueError: Table 'vitals' has unknown reserved key(s) ['_default']. Reserved keys at the
table level: '_defaults', '_table'. Event names may not begin with an underscore.
>>> TableConfig.parse("vitals", {
...     "_tables": {"join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}}},
...     "hr": {"code": "HR", "time": None},
... })
Traceback (most recent call last):
    ...
ValueError: Table 'vitals' has unknown reserved key(s) ['_tables']. Reserved keys at the
table level: '_defaults', '_table'. Event names may not begin with an underscore.

Source code in MEDS_extract/config.py

@dataclass(frozen=True)
class TableConfig:
    """Fully-resolved config for one source table block.

    Global ``_defaults`` are merged with file-level ``_defaults`` at parse time,
    so :attr:`subject_id_node` already reflects the final inherited value.
    ``_table`` sub-keys (``cols``, ``join``) are lifted to top-level fields.

    Examples:
        >>> tc = TableConfig.parse("patients", {
        ...     "_defaults": {"subject_id": "$MRN"},
        ...     "_table": {"join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}}},
        ...     "dob": {"code": "BIRTH", "time": "$dob"},
        ... })
        >>> tc.input_prefix
        'patients'
        >>> type(tc.subject_id_node).__name__
        'Column'
        >>> sorted(tc.subject_id_node.referenced_columns)
        ['MRN']
        >>> tc.join.input_prefix
        'stays'
        >>> [e.name for e in tc.events]
        ['dob']

        ``_defaults`` accepts only the keys it actually consumes — a stray key
        would otherwise be silently dropped. The sharpest case is ``time``: an
        event with no ``time`` key is legally static, so a dropped default
        ``time`` would turn every event in the table static without a word:

        >>> TableConfig.parse("vitals", {
        ...     "_defaults": {"subject_id": "$MRN", "time": '$charttime::"%Y-%m-%d"'},
        ...     "hr": {"code": "HR", "numeric_value": "$hr"},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: Table 'vitals' has unknown keys under '_defaults': ['time']. Allowed keys:
        'subject_id'.

        A typo'd ``subject_id`` is caught the same way, instead of silently
        falling back to reading a literal ``subject_id`` source column:

        >>> TableConfig.parse("vitals", {
        ...     "_defaults": {"subject_ids": "$MRN"},
        ...     "hr": {"code": "HR", "time": None},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: Table 'vitals' has unknown keys under '_defaults': ['subject_ids']. Allowed
        keys: 'subject_id'.

        Underscore-prefixed keys other than ``_defaults``/``_table`` are almost
        certainly typos of those reserved names, and are rejected up front rather
        than falling through to event parsing with a misleading error:

        >>> TableConfig.parse("vitals", {
        ...     "_default": {"subject_id": "$MRN"},
        ...     "hr": {"code": "HR", "time": None},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: Table 'vitals' has unknown reserved key(s) ['_default']. Reserved keys at the
        table level: '_defaults', '_table'. Event names may not begin with an underscore.
        >>> TableConfig.parse("vitals", {
        ...     "_tables": {"join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}}},
        ...     "hr": {"code": "HR", "time": None},
        ... })
        Traceback (most recent call last):
            ...
        ValueError: Table 'vitals' has unknown reserved key(s) ['_tables']. Reserved keys at the
        table level: '_defaults', '_table'. Event names may not begin with an underscore.
    """

    input_prefix: str
    subject_id_node: NodeBase | None = None
    cols: dict[str, NodeBase] = field(default_factory=dict)
    join: JoinConfig | None = None
    events: tuple[EventConfig, ...] = ()

    def __post_init__(self):
        if not self.events:
            raise ValueError(
                f"Table '{self.input_prefix}' defines no events. Every table must contain at "
                f"least one event (a dict with at minimum a 'code' key)."
            )
        if self.subject_id_node is not None and not isinstance(self.subject_id_node, NodeBase):
            raise TypeError(
                f"Table '{self.input_prefix}' subject_id_node must be a parsed dftly node, "
                f"got {type(self.subject_id_node).__name__}."
            )
        for k, v in self.cols.items():
            if not isinstance(v, NodeBase):
                raise TypeError(
                    f"Table '{self.input_prefix}' derived column '{k}' must be a parsed dftly "
                    f"node, got {type(v).__name__}."
                )
        for event in self.events:
            if self.input_prefix in event.metadata:
                logger.warning(
                    f"Table '{self.input_prefix}' event '{event.name}': its _metadata block "
                    f"targets the event's own source table. extract_code_metadata will re-scan "
                    f"the raw table, fully materialize the mapped frame, and join it against "
                    f"every observed code — on a large table this exhausts memory. Use the "
                    f"'{SELF_METADATA_PREFIX}' prefix instead to source these columns from the "
                    f"event's own extracted rows."
                )
        if self.join is not None:
            # A join key may be a ``_table.cols`` derived column (the canonical case: a
            # literal restricting the join to matching right rows). Such an entry is
            # materialized BEFORE the join runs — earlier than the rest of the derived
            # columns — so it must be self-contained: computable from raw source columns
            # alone, without referencing other derived names (including its own).
            for k in self.join.left_on:
                if k in self.cols:
                    tainted = sorted(set(self.cols[k].referenced_columns) & set(self.cols))
                    if tainted:
                        raise ValueError(
                            f"Table '{self.input_prefix}' join key '{k}' is a derived column that "
                            f"references other derived column(s) {tainted}. Join-key derived "
                            f"columns are applied before the join and must be computable from "
                            f"raw source columns alone."
                        )
            # A self-join can never deliver its columns: every column of the joined
            # table exists on the left by construction (same file), so every output
            # would collide. Rejected at parse so the failure names the pattern
            # instead of surfacing per-column at scan time.
            if self.join.input_prefix == self.input_prefix:
                raise ValueError(
                    f"Table '{self.input_prefix}' joins to itself. A self-join cannot work: "
                    f"every joined column already exists on the left side (same file), and "
                    f"MESSY does not support suffixed join outputs. Compute per-group "
                    f"reductions of a table's own columns in a pre-processing step instead."
                )
            # A reference to '<col>_right' means the config was written against
            # polars' collision suffix, which MESSY does not support.
            refs: set[str] = set()
            if self.subject_id_node is not None:
                refs.update(self.subject_id_node.referenced_columns)
            for node in self.cols.values():
                refs.update(node.referenced_columns)
            for event in self.events:
                refs.update(event.referenced_columns)
            suffixed = sorted(
                r for r in refs if r.endswith("_right") and r[: -len("_right")] in self.join.cols
            )
            if suffixed:
                raise ValueError(
                    f"Table '{self.input_prefix}' references column(s) {suffixed}, the "
                    f"'_right'-suffixed form of joined column(s). Suffixed join outputs are "
                    f"not supported: joined columns must arrive under names the left table "
                    f"does not already have, referenced without a suffix. Rename the "
                    f"colliding source column upstream, or compute the value in a "
                    f"pre-processing step."
                )
            # Derived expressions are applied AFTER the join (except join keys), so a
            # derived column sharing a joined column's name would silently overwrite
            # the joined values. Same-named join keys are exempt: the join coalesces
            # them, and re-deriving a self-contained key is a no-op.
            coalesced = {
                lft for lft, r in zip(self.join.left_on, self.join.right_on, strict=False) if lft == r
            }
            clobbered = sorted((set(self.cols) & set(self.join.cols)) - coalesced)
            if clobbered:
                raise ValueError(
                    f"Table '{self.input_prefix}': derived column(s) {clobbered} under "
                    f"'_table.cols' share names with columns delivered by the join. Derived "
                    f"expressions are applied after the join and would silently overwrite the "
                    f"joined values. Give the derived column and the joined column distinct "
                    f"names."
                )

    @classmethod
    def parse(
        cls,
        input_prefix: str,
        raw: Mapping[str, Any],
        global_defaults: Mapping[str, Any] | None = None,
    ) -> TableConfig:
        raw = dict(raw)
        global_defaults = dict(global_defaults or {})

        file_defaults = dict(raw.pop("_defaults", {}))
        merged_defaults = {**global_defaults, **file_defaults}
        unknown_default_keys = set(merged_defaults) - {"subject_id"}
        if unknown_default_keys:
            raise ValueError(
                f"Table '{input_prefix}' has unknown keys under '_defaults': "
                f"{sorted(unknown_default_keys)}. Allowed keys: 'subject_id'."
            )

        table_cfg = dict(raw.pop("_table", {}))
        unknown_table_keys = set(table_cfg) - {"cols", "join"}
        if unknown_table_keys:
            raise ValueError(
                f"Table '{input_prefix}' has unknown keys under '_table': "
                f"{sorted(unknown_table_keys)}. Allowed keys: 'cols', 'join'."
            )

        stray_reserved = sorted(k for k in raw if k.startswith("_"))
        if stray_reserved:
            raise ValueError(
                f"Table '{input_prefix}' has unknown reserved key(s) {stray_reserved}. Reserved "
                f"keys at the table level: '_defaults', '_table'. Event names may not begin with "
                f"an underscore."
            )
        parser = Parser()

        raw_cols = dict(table_cfg.get("cols", {}))
        cols = {k: v if isinstance(v, NodeBase) else parser(v) for k, v in raw_cols.items()}

        join = JoinConfig.parse(dict(table_cfg["join"])) if "join" in table_cfg else None

        subject_id_raw = merged_defaults.get("subject_id")
        subject_id_node: NodeBase | None
        if subject_id_raw is None:
            subject_id_node = None
        elif isinstance(subject_id_raw, NodeBase):
            subject_id_node = subject_id_raw
        else:
            subject_id_node = parser(subject_id_raw)

        events = tuple(EventConfig.parse(event_name, event_raw) for event_name, event_raw in raw.items())

        return cls(
            input_prefix=input_prefix,
            subject_id_node=subject_id_node,
            cols=cols,
            join=join,
            events=events,
        )

    @cached_property
    def subject_id_polars_expr(self) -> pl.Expr:
        """Polars expression producing the MEDS ``subject_id`` column.

        Always produces ``Int64`` output, per the MEDS schema. When no explicit
        subject_id expression is configured, the expression reads the existing
        ``subject_id`` column from the source table and casts it to Int64.
        Non-hash expressions are cast via :meth:`polars.Expr.cast` in **strict
        mode** — values that can't be converted (e.g., unparsable strings)
        raise at query time rather than silently becoming nulls. ``hash()``
        outputs (UInt64) are reinterpreted to preserve bits rather than cast —
        a cast would null (or raise on) every value above ``i64.max``, roughly
        half the hash space. The consequence to know: the result is a
        bit-reinterpret of polars' hash, not a fresh signed hash, so an
        external system computing its own hashes will not get bit-compatible
        subject IDs.

        The returned expression is never ``None`` — stages can apply it
        unconditionally.

        Examples:
            No explicit subject_id expression: read the existing ``subject_id``
            column (cast to Int64 for schema compliance).

            >>> import polars as pl
            >>> tc = TableConfig.parse("t", {"e": {"code": "X", "time": None}})
            >>> df = pl.DataFrame({"subject_id": [1, 2]}, schema={"subject_id": pl.Int32})
            >>> df.select(subject_id=tc.subject_id_polars_expr).schema["subject_id"]
            Int64

            Column-reference expression: Int32 source → Int64 output.

            >>> tc = TableConfig.parse("t", {"_defaults": {"subject_id": "$patient_id"},
            ...                              "e": {"code": "X", "time": None}})
            >>> df = pl.DataFrame({"patient_id": [1, 2]}, schema={"patient_id": pl.Int32})
            >>> df.select(subject_id=tc.subject_id_polars_expr).schema["subject_id"]
            Int64

            ``hash()`` produces UInt64; we reinterpret to Int64.

            >>> tc = TableConfig.parse("t", {"_defaults": {"subject_id": "hash($mrn)"},
            ...                              "e": {"code": "X", "time": None}})
            >>> df = pl.DataFrame({"mrn": ["ABC", "DEF"]})
            >>> df.select(subject_id=tc.subject_id_polars_expr).schema["subject_id"]
            Int64
        """
        if self.subject_id_node is None:
            return pl.col("subject_id").cast(pl.Int64, strict=True)
        expr = self.subject_id_node.polars_expr
        if isinstance(self.subject_id_node, Hash):
            return expr.reinterpret(signed=True)
        return expr.cast(pl.Int64, strict=True)

    @property
    def col_outputs(self) -> set[str]:
        """Column names produced by ``_table.cols`` — derived, not read from source."""
        return set(self.cols.keys())

    @property
    def joined_columns(self) -> set[str]:
        """Column names that come from the joined table, not the source file."""
        return set(self.join.cols) if self.join is not None else set()

    def source_columns(self) -> set[str]:
        """Columns that must be read from this table's source parquet file.

        Aggregates the subject_id source columns, the join left key, derived-column
        inputs, and all event-referenced columns — minus columns produced by
        ``_table.cols`` (derived) or pulled in by the join (come from elsewhere).

        Examples:
            >>> tc = TableConfig.parse("labs", {
            ...     "_defaults": {"subject_id": "$patient_id"},
            ...     "_table": {
            ...         "cols": {"year": "$anchor_year - $anchor_age"},
            ...         "join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}},
            ...     },
            ...     "lab": {"code": "$test", "time": "$dischtime"},
            ... })
            >>> sorted(tc.source_columns())
            ['anchor_age', 'anchor_year', 'patient_id', 'stay_id', 'test']
        """
        cols: set[str] = set()
        if self.subject_id_node is not None:
            cols.update(self.subject_id_node.referenced_columns)
        else:
            cols.add("subject_id")

        for node in self.cols.values():
            cols.update(node.referenced_columns)

        if self.join is not None:
            cols.update(self.join.left_on)

        for event in self.events:
            cols.update(event.referenced_columns)

        return cols - self.col_outputs - self.joined_columns

    def source_files(self, dir: Path | UPath) -> list[Path | UPath]:
        """Resolve the list of source files for this table under ``dir``.

        Delegates to :func:`MEDS_extract.io.resolve_source_files`. Returns a
        non-empty list; raises ``FileNotFoundError`` or ``ValueError`` (on
        layout ambiguity) otherwise.
        """
        return resolve_source_files(dir, self.input_prefix)

    def scan(self, dir: Path | UPath) -> pl.LazyFrame:
        """Scan every source file for this table under ``dir``, apply the join.

        The unified entry point that every stage should use to read a table's data. Auto-detects the layout
        (bare file vs sub-sharded directory), dispatches on format, and applies the join if configured.
        """
        df = scan_source(self.source_files(dir))
        return self.apply_join(df, dir)

    def apply_join(self, df: pl.LazyFrame, input_dir: Path | UPath) -> pl.LazyFrame:
        """Apply this table's join to an already-scanned frame (no-op without a join).

        Any ``_table.cols`` entry named as a left join key is materialized first: a
        left key may be a derived column — canonically a literal that restricts the
        join to matching right rows, e.g. joining on ``[pharmacy_id, drug_type]``
        with ``cols: {drug_type: "'MAIN'"}`` to pull only the MAIN-row NDC. Such
        entries are validated at parse time to be computable from raw source columns
        alone, and re-deriving them later in :meth:`prepare` is a no-op by that same
        self-containment.

        Examples:
            >>> tc = TableConfig.parse("emar", {
            ...     "_defaults": {"subject_id": "$subject_id"},
            ...     "_table": {
            ...         "cols": {"drug_type": "'MAIN'"},
            ...         "join": {"prescriptions": {
            ...             "key": ["pharmacy_id", "drug_type"], "cols": ["ndc"],
            ...         }},
            ...     },
            ...     "med": {"code": 'f"MED//{$ndc}"', "time": None},
            ... })
            >>> with yaml_disk('''
            ... prescriptions.parquet:
            ...   pharmacy_id: [1, 1, 2]
            ...   drug_type: ["MAIN", "BASE", "MAIN"]
            ...   ndc: ["10", "0", "20"]
            ... ''') as d:
            ...     emar = pl.LazyFrame({"subject_id": [1, 2], "pharmacy_id": [1, 2]})
            ...     tc.apply_join(emar, Path(d)).sort("subject_id").collect()
            shape: (2, 4)
            ┌────────────┬─────────────┬───────────┬─────┐
            │ subject_id ┆ pharmacy_id ┆ drug_type ┆ ndc │
            │ ---        ┆ ---         ┆ ---       ┆ --- │
            │ i64        ┆ i64         ┆ str       ┆ str │
            ╞════════════╪═════════════╪═══════════╪═════╡
            │ 1          ┆ 1           ┆ MAIN      ┆ 10  │
            │ 2          ┆ 2           ┆ MAIN      ┆ 20  │
            └────────────┴─────────────┴───────────┴─────┘

            One row per input row — the BASE sibling of pharmacy 1 is excluded by the
            composite key rather than fanning the join out.
        """
        if self.join is None:
            return df
        prejoin = [k for k in self.join.left_on if k in self.cols]
        if prejoin:
            df = df.with_columns(*(self.cols[k].polars_expr.alias(k) for k in prejoin))
        return self.join.apply(df, input_dir)

    def prepare(self, df: pl.LazyFrame) -> pl.LazyFrame:
        """Apply subject_id materialization and derived columns to a raw dataframe.

        The result has an ``Int64`` ``subject_id`` column and any ``_table.cols``
        derived columns. It does **not** apply the join — joins need a
        stage-specific input directory for the join target and are applied by
        the caller via :meth:`JoinConfig.apply`.

        Derived columns are applied in insertion order via a chain of
        ``with_columns`` calls, so a later entry may reference a name defined
        above it in the same ``_table.cols`` block. A forward reference (a
        column referencing a name defined below it) fails at query time with
        polars' standard missing-column error — intentionally not special-cased.

        Examples:
            >>> _ = pl.Config.set_tbl_width_chars(600)
            >>> tc = TableConfig.parse("t", {
            ...     "_defaults": {"subject_id": "$MRN"},
            ...     "_table": {"cols": {"year": "$anchor_year - $anchor_age"}},
            ...     "e": {"code": "X", "time": None},
            ... })
            >>> raw = pl.DataFrame({
            ...     "MRN": [100, 200],
            ...     "anchor_year": [2020, 2021],
            ...     "anchor_age": [30, 25],
            ... })
            >>> tc.prepare(raw.lazy()).collect().select("subject_id", "year")
            shape: (2, 2)
            ┌────────────┬──────┐
            │ subject_id ┆ year │
            │ ---        ┆ ---  │
            │ i64        ┆ i64  │
            ╞════════════╪══════╡
            │ 100        ┆ 1990 │
            │ 200        ┆ 1996 │
            └────────────┴──────┘

            Later ``_table.cols`` entries can reference earlier ones — the
            motivating case is offset-time schemas (eICU, HIRID, ...) where
            pseudotimes chain off each other. Here ``age_at_admit`` uses the
            ``year_of_birth`` defined on the line above:

            >>> tc = TableConfig.parse("t", {
            ...     "_defaults": {"subject_id": "$MRN"},
            ...     "_table": {"cols": {
            ...         "year_of_birth": "$anchor_year - $anchor_age",
            ...         "age_at_admit":  "$admit_year - $year_of_birth",
            ...     }},
            ...     "e": {"code": "X", "time": None},
            ... })
            >>> raw = pl.DataFrame({
            ...     "MRN": [100, 200],
            ...     "anchor_year": [2020, 2021],
            ...     "anchor_age": [30, 25],
            ...     "admit_year": [2024, 2024],
            ... })
            >>> tc.prepare(raw.lazy()).collect().select(
            ...     "subject_id", "year_of_birth", "age_at_admit"
            ... )
            shape: (2, 3)
            ┌────────────┬───────────────┬──────────────┐
            │ subject_id ┆ year_of_birth ┆ age_at_admit │
            │ ---        ┆ ---           ┆ ---          │
            │ i64        ┆ i64           ┆ i64          │
            ╞════════════╪═══════════════╪══════════════╡
            │ 100        ┆ 1990          ┆ 34           │
            │ 200        ┆ 1996          ┆ 28           │
            └────────────┴───────────────┴──────────────┘
        """
        df = df.with_columns(subject_id=self.subject_id_polars_expr)
        for name, node in self.cols.items():
            df = df.with_columns(node.polars_expr.alias(name))
        return df

    def extract_events(
        self,
        df: pl.LazyFrame,
        do_dedup_text_and_numeric: bool = False,
    ) -> pl.LazyFrame:
        """Prepare ``df`` and extract every event in this table, concatenated.

        Each event's output rows are tagged with a ``source_block`` column derived
        from ``f"{input_prefix}/{event.name}"``.

        Raises:
            ValueError: if extracting any individual event fails (the table + event
                name are included in the error).

        Examples:
            >>> _ = pl.Config.set_tbl_width_chars(600)
            >>> tc = TableConfig.parse("data", {
            ...     "admit": {"code": 'f"ADMIT//{$dept}"', "time": '$ts::"%Y-%m-%d"'},
            ...     "color": {"code": "EYE_COLOR", "time": None, "eye_color": "$color"},
            ... })
            >>> raw = pl.DataFrame({
            ...     "subject_id": [1, 2],
            ...     "dept": ["CARDIAC", "PULM"],
            ...     "ts": ["2021-01-01", "2021-01-02"],
            ...     "color": ["blue", "green"],
            ... })
            >>> tc.extract_events(raw.lazy()).collect().select(
            ...     "subject_id", "code", "source_block"
            ... ).sort("source_block", "subject_id")
            shape: (4, 3)
            ┌────────────┬────────────────┬──────────────┐
            │ subject_id ┆ code           ┆ source_block │
            │ ---        ┆ ---            ┆ ---          │
            │ i64        ┆ str            ┆ str          │
            ╞════════════╪════════════════╪══════════════╡
            │ 1          ┆ ADMIT//CARDIAC ┆ data/admit   │
            │ 2          ┆ ADMIT//PULM    ┆ data/admit   │
            │ 1          ┆ EYE_COLOR      ┆ data/color   │
            │ 2          ┆ EYE_COLOR      ┆ data/color   │
            └────────────┴────────────────┴──────────────┘
        """
        df = self.prepare(df)

        event_dfs = []
        for event in self.events:
            source_block = f"{self.input_prefix}/{event.name}"
            try:
                logger.info(f"Building extraction plan for {source_block}")
                event_dfs.append(
                    event.extract(
                        df, source_block=source_block, do_dedup_text_and_numeric=do_dedup_text_and_numeric
                    )
                )
            except Exception as e:
                raise ValueError(f"Error extracting event {source_block}: {e}") from e

        return pl.concat(event_dfs, how="diagonal_relaxed")

col_outputs property

Column names produced by _table.cols — derived, not read from source.

joined_columns property

Column names that come from the joined table, not the source file.

subject_id_polars_expr cached property

Polars expression producing the MEDS subject_id column.

Always produces Int64 output, per the MEDS schema. When no explicit subject_id expression is configured, the expression reads the existing subject_id column from the source table and casts it to Int64. Non-hash expressions are cast via :meth:polars.Expr.cast in strict mode — values that can’t be converted (e.g., unparsable strings) raise at query time rather than silently becoming nulls. hash() outputs (UInt64) are reinterpreted to preserve bits rather than cast — a cast would null (or raise on) every value above i64.max, roughly half the hash space. The consequence to know: the result is a bit-reinterpret of polars’ hash, not a fresh signed hash, so an external system computing its own hashes will not get bit-compatible subject IDs.

The returned expression is never None — stages can apply it unconditionally.

Examples:

No explicit subject_id expression: read the existing subject_id column (cast to Int64 for schema compliance).

>>> import polars as pl
>>> tc = TableConfig.parse("t", {"e": {"code": "X", "time": None}})
>>> df = pl.DataFrame({"subject_id": [1, 2]}, schema={"subject_id": pl.Int32})
>>> df.select(subject_id=tc.subject_id_polars_expr).schema["subject_id"]
Int64

Column-reference expression: Int32 source → Int64 output.

>>> tc = TableConfig.parse("t", {"_defaults": {"subject_id": "$patient_id"},
...                              "e": {"code": "X", "time": None}})
>>> df = pl.DataFrame({"patient_id": [1, 2]}, schema={"patient_id": pl.Int32})
>>> df.select(subject_id=tc.subject_id_polars_expr).schema["subject_id"]
Int64

hash() produces UInt64; we reinterpret to Int64.

>>> tc = TableConfig.parse("t", {"_defaults": {"subject_id": "hash($mrn)"},
...                              "e": {"code": "X", "time": None}})
>>> df = pl.DataFrame({"mrn": ["ABC", "DEF"]})
>>> df.select(subject_id=tc.subject_id_polars_expr).schema["subject_id"]
Int64

apply_join(df, input_dir)

Apply this table’s join to an already-scanned frame (no-op without a join).

Any _table.cols entry named as a left join key is materialized first: a left key may be a derived column — canonically a literal that restricts the join to matching right rows, e.g. joining on [pharmacy_id, drug_type] with cols: {drug_type: "'MAIN'"} to pull only the MAIN-row NDC. Such entries are validated at parse time to be computable from raw source columns alone, and re-deriving them later in :meth:prepare is a no-op by that same self-containment.

Examples:

>>> tc = TableConfig.parse("emar", {
...     "_defaults": {"subject_id": "$subject_id"},
...     "_table": {
...         "cols": {"drug_type": "'MAIN'"},
...         "join": {"prescriptions": {
...             "key": ["pharmacy_id", "drug_type"], "cols": ["ndc"],
...         }},
...     },
...     "med": {"code": 'f"MED//{$ndc}"', "time": None},
... })
>>> with yaml_disk('''
... prescriptions.parquet:
...   pharmacy_id: [1, 1, 2]
...   drug_type: ["MAIN", "BASE", "MAIN"]
...   ndc: ["10", "0", "20"]
... ''') as d:
...     emar = pl.LazyFrame({"subject_id": [1, 2], "pharmacy_id": [1, 2]})
...     tc.apply_join(emar, Path(d)).sort("subject_id").collect()
shape: (2, 4)
┌────────────┬─────────────┬───────────┬─────┐
│ subject_id ┆ pharmacy_id ┆ drug_type ┆ ndc │
│ ---        ┆ ---         ┆ ---       ┆ --- │
│ i64        ┆ i64         ┆ str       ┆ str │
╞════════════╪═════════════╪═══════════╪═════╡
│ 1          ┆ 1           ┆ MAIN      ┆ 10  │
│ 2          ┆ 2           ┆ MAIN      ┆ 20  │
└────────────┴─────────────┴───────────┴─────┘

One row per input row — the BASE sibling of pharmacy 1 is excluded by the composite key rather than fanning the join out.

Source code in MEDS_extract/config.py

def apply_join(self, df: pl.LazyFrame, input_dir: Path | UPath) -> pl.LazyFrame:
    """Apply this table's join to an already-scanned frame (no-op without a join).

    Any ``_table.cols`` entry named as a left join key is materialized first: a
    left key may be a derived column — canonically a literal that restricts the
    join to matching right rows, e.g. joining on ``[pharmacy_id, drug_type]``
    with ``cols: {drug_type: "'MAIN'"}`` to pull only the MAIN-row NDC. Such
    entries are validated at parse time to be computable from raw source columns
    alone, and re-deriving them later in :meth:`prepare` is a no-op by that same
    self-containment.

    Examples:
        >>> tc = TableConfig.parse("emar", {
        ...     "_defaults": {"subject_id": "$subject_id"},
        ...     "_table": {
        ...         "cols": {"drug_type": "'MAIN'"},
        ...         "join": {"prescriptions": {
        ...             "key": ["pharmacy_id", "drug_type"], "cols": ["ndc"],
        ...         }},
        ...     },
        ...     "med": {"code": 'f"MED//{$ndc}"', "time": None},
        ... })
        >>> with yaml_disk('''
        ... prescriptions.parquet:
        ...   pharmacy_id: [1, 1, 2]
        ...   drug_type: ["MAIN", "BASE", "MAIN"]
        ...   ndc: ["10", "0", "20"]
        ... ''') as d:
        ...     emar = pl.LazyFrame({"subject_id": [1, 2], "pharmacy_id": [1, 2]})
        ...     tc.apply_join(emar, Path(d)).sort("subject_id").collect()
        shape: (2, 4)
        ┌────────────┬─────────────┬───────────┬─────┐
        │ subject_id ┆ pharmacy_id ┆ drug_type ┆ ndc │
        │ ---        ┆ ---         ┆ ---       ┆ --- │
        │ i64        ┆ i64         ┆ str       ┆ str │
        ╞════════════╪═════════════╪═══════════╪═════╡
        │ 1          ┆ 1           ┆ MAIN      ┆ 10  │
        │ 2          ┆ 2           ┆ MAIN      ┆ 20  │
        └────────────┴─────────────┴───────────┴─────┘

        One row per input row — the BASE sibling of pharmacy 1 is excluded by the
        composite key rather than fanning the join out.
    """
    if self.join is None:
        return df
    prejoin = [k for k in self.join.left_on if k in self.cols]
    if prejoin:
        df = df.with_columns(*(self.cols[k].polars_expr.alias(k) for k in prejoin))
    return self.join.apply(df, input_dir)

extract_events(df, do_dedup_text_and_numeric=False)

Prepare df and extract every event in this table, concatenated.

Each event’s output rows are tagged with a source_block column derived from f"{input_prefix}/{event.name}".

Raises:

Type	Description
ValueError	if extracting any individual event fails (the table + event name are included in the error).

Examples:

>>> _ = pl.Config.set_tbl_width_chars(600)
>>> tc = TableConfig.parse("data", {
...     "admit": {"code": 'f"ADMIT//{$dept}"', "time": '$ts::"%Y-%m-%d"'},
...     "color": {"code": "EYE_COLOR", "time": None, "eye_color": "$color"},
... })
>>> raw = pl.DataFrame({
...     "subject_id": [1, 2],
...     "dept": ["CARDIAC", "PULM"],
...     "ts": ["2021-01-01", "2021-01-02"],
...     "color": ["blue", "green"],
... })
>>> tc.extract_events(raw.lazy()).collect().select(
...     "subject_id", "code", "source_block"
... ).sort("source_block", "subject_id")
shape: (4, 3)
┌────────────┬────────────────┬──────────────┐
│ subject_id ┆ code           ┆ source_block │
│ ---        ┆ ---            ┆ ---          │
│ i64        ┆ str            ┆ str          │
╞════════════╪════════════════╪══════════════╡
│ 1          ┆ ADMIT//CARDIAC ┆ data/admit   │
│ 2          ┆ ADMIT//PULM    ┆ data/admit   │
│ 1          ┆ EYE_COLOR      ┆ data/color   │
│ 2          ┆ EYE_COLOR      ┆ data/color   │
└────────────┴────────────────┴──────────────┘

Source code in MEDS_extract/config.py

def extract_events(
    self,
    df: pl.LazyFrame,
    do_dedup_text_and_numeric: bool = False,
) -> pl.LazyFrame:
    """Prepare ``df`` and extract every event in this table, concatenated.

    Each event's output rows are tagged with a ``source_block`` column derived
    from ``f"{input_prefix}/{event.name}"``.

    Raises:
        ValueError: if extracting any individual event fails (the table + event
            name are included in the error).

    Examples:
        >>> _ = pl.Config.set_tbl_width_chars(600)
        >>> tc = TableConfig.parse("data", {
        ...     "admit": {"code": 'f"ADMIT//{$dept}"', "time": '$ts::"%Y-%m-%d"'},
        ...     "color": {"code": "EYE_COLOR", "time": None, "eye_color": "$color"},
        ... })
        >>> raw = pl.DataFrame({
        ...     "subject_id": [1, 2],
        ...     "dept": ["CARDIAC", "PULM"],
        ...     "ts": ["2021-01-01", "2021-01-02"],
        ...     "color": ["blue", "green"],
        ... })
        >>> tc.extract_events(raw.lazy()).collect().select(
        ...     "subject_id", "code", "source_block"
        ... ).sort("source_block", "subject_id")
        shape: (4, 3)
        ┌────────────┬────────────────┬──────────────┐
        │ subject_id ┆ code           ┆ source_block │
        │ ---        ┆ ---            ┆ ---          │
        │ i64        ┆ str            ┆ str          │
        ╞════════════╪════════════════╪══════════════╡
        │ 1          ┆ ADMIT//CARDIAC ┆ data/admit   │
        │ 2          ┆ ADMIT//PULM    ┆ data/admit   │
        │ 1          ┆ EYE_COLOR      ┆ data/color   │
        │ 2          ┆ EYE_COLOR      ┆ data/color   │
        └────────────┴────────────────┴──────────────┘
    """
    df = self.prepare(df)

    event_dfs = []
    for event in self.events:
        source_block = f"{self.input_prefix}/{event.name}"
        try:
            logger.info(f"Building extraction plan for {source_block}")
            event_dfs.append(
                event.extract(
                    df, source_block=source_block, do_dedup_text_and_numeric=do_dedup_text_and_numeric
                )
            )
        except Exception as e:
            raise ValueError(f"Error extracting event {source_block}: {e}") from e

    return pl.concat(event_dfs, how="diagonal_relaxed")

prepare(df)

Apply subject_id materialization and derived columns to a raw dataframe.

The result has an Int64 subject_id column and any _table.cols derived columns. It does not apply the join — joins need a stage-specific input directory for the join target and are applied by the caller via :meth:JoinConfig.apply.

Derived columns are applied in insertion order via a chain of with_columns calls, so a later entry may reference a name defined above it in the same _table.cols block. A forward reference (a column referencing a name defined below it) fails at query time with polars’ standard missing-column error — intentionally not special-cased.

Examples:

>>> _ = pl.Config.set_tbl_width_chars(600)
>>> tc = TableConfig.parse("t", {
...     "_defaults": {"subject_id": "$MRN"},
...     "_table": {"cols": {"year": "$anchor_year - $anchor_age"}},
...     "e": {"code": "X", "time": None},
... })
>>> raw = pl.DataFrame({
...     "MRN": [100, 200],
...     "anchor_year": [2020, 2021],
...     "anchor_age": [30, 25],
... })
>>> tc.prepare(raw.lazy()).collect().select("subject_id", "year")
shape: (2, 2)
┌────────────┬──────┐
│ subject_id ┆ year │
│ ---        ┆ ---  │
│ i64        ┆ i64  │
╞════════════╪══════╡
│ 100        ┆ 1990 │
│ 200        ┆ 1996 │
└────────────┴──────┘

Later _table.cols entries can reference earlier ones — the motivating case is offset-time schemas (eICU, HIRID, …) where pseudotimes chain off each other. Here age_at_admit uses the year_of_birth defined on the line above:

>>> tc = TableConfig.parse("t", {
...     "_defaults": {"subject_id": "$MRN"},
...     "_table": {"cols": {
...         "year_of_birth": "$anchor_year - $anchor_age",
...         "age_at_admit":  "$admit_year - $year_of_birth",
...     }},
...     "e": {"code": "X", "time": None},
... })
>>> raw = pl.DataFrame({
...     "MRN": [100, 200],
...     "anchor_year": [2020, 2021],
...     "anchor_age": [30, 25],
...     "admit_year": [2024, 2024],
... })
>>> tc.prepare(raw.lazy()).collect().select(
...     "subject_id", "year_of_birth", "age_at_admit"
... )
shape: (2, 3)
┌────────────┬───────────────┬──────────────┐
│ subject_id ┆ year_of_birth ┆ age_at_admit │
│ ---        ┆ ---           ┆ ---          │
│ i64        ┆ i64           ┆ i64          │
╞════════════╪═══════════════╪══════════════╡
│ 100        ┆ 1990          ┆ 34           │
│ 200        ┆ 1996          ┆ 28           │
└────────────┴───────────────┴──────────────┘

Source code in MEDS_extract/config.py

def prepare(self, df: pl.LazyFrame) -> pl.LazyFrame:
    """Apply subject_id materialization and derived columns to a raw dataframe.

    The result has an ``Int64`` ``subject_id`` column and any ``_table.cols``
    derived columns. It does **not** apply the join — joins need a
    stage-specific input directory for the join target and are applied by
    the caller via :meth:`JoinConfig.apply`.

    Derived columns are applied in insertion order via a chain of
    ``with_columns`` calls, so a later entry may reference a name defined
    above it in the same ``_table.cols`` block. A forward reference (a
    column referencing a name defined below it) fails at query time with
    polars' standard missing-column error — intentionally not special-cased.

    Examples:
        >>> _ = pl.Config.set_tbl_width_chars(600)
        >>> tc = TableConfig.parse("t", {
        ...     "_defaults": {"subject_id": "$MRN"},
        ...     "_table": {"cols": {"year": "$anchor_year - $anchor_age"}},
        ...     "e": {"code": "X", "time": None},
        ... })
        >>> raw = pl.DataFrame({
        ...     "MRN": [100, 200],
        ...     "anchor_year": [2020, 2021],
        ...     "anchor_age": [30, 25],
        ... })
        >>> tc.prepare(raw.lazy()).collect().select("subject_id", "year")
        shape: (2, 2)
        ┌────────────┬──────┐
        │ subject_id ┆ year │
        │ ---        ┆ ---  │
        │ i64        ┆ i64  │
        ╞════════════╪══════╡
        │ 100        ┆ 1990 │
        │ 200        ┆ 1996 │
        └────────────┴──────┘

        Later ``_table.cols`` entries can reference earlier ones — the
        motivating case is offset-time schemas (eICU, HIRID, ...) where
        pseudotimes chain off each other. Here ``age_at_admit`` uses the
        ``year_of_birth`` defined on the line above:

        >>> tc = TableConfig.parse("t", {
        ...     "_defaults": {"subject_id": "$MRN"},
        ...     "_table": {"cols": {
        ...         "year_of_birth": "$anchor_year - $anchor_age",
        ...         "age_at_admit":  "$admit_year - $year_of_birth",
        ...     }},
        ...     "e": {"code": "X", "time": None},
        ... })
        >>> raw = pl.DataFrame({
        ...     "MRN": [100, 200],
        ...     "anchor_year": [2020, 2021],
        ...     "anchor_age": [30, 25],
        ...     "admit_year": [2024, 2024],
        ... })
        >>> tc.prepare(raw.lazy()).collect().select(
        ...     "subject_id", "year_of_birth", "age_at_admit"
        ... )
        shape: (2, 3)
        ┌────────────┬───────────────┬──────────────┐
        │ subject_id ┆ year_of_birth ┆ age_at_admit │
        │ ---        ┆ ---           ┆ ---          │
        │ i64        ┆ i64           ┆ i64          │
        ╞════════════╪═══════════════╪══════════════╡
        │ 100        ┆ 1990          ┆ 34           │
        │ 200        ┆ 1996          ┆ 28           │
        └────────────┴───────────────┴──────────────┘
    """
    df = df.with_columns(subject_id=self.subject_id_polars_expr)
    for name, node in self.cols.items():
        df = df.with_columns(node.polars_expr.alias(name))
    return df

scan(dir)

Scan every source file for this table under dir, apply the join.

The unified entry point that every stage should use to read a table’s data. Auto-detects the layout (bare file vs sub-sharded directory), dispatches on format, and applies the join if configured.

Source code in MEDS_extract/config.py

def scan(self, dir: Path | UPath) -> pl.LazyFrame:
    """Scan every source file for this table under ``dir``, apply the join.

    The unified entry point that every stage should use to read a table's data. Auto-detects the layout
    (bare file vs sub-sharded directory), dispatches on format, and applies the join if configured.
    """
    df = scan_source(self.source_files(dir))
    return self.apply_join(df, dir)

source_columns()

Columns that must be read from this table’s source parquet file.

Aggregates the subject_id source columns, the join left key, derived-column inputs, and all event-referenced columns — minus columns produced by _table.cols (derived) or pulled in by the join (come from elsewhere).

Examples:

>>> tc = TableConfig.parse("labs", {
...     "_defaults": {"subject_id": "$patient_id"},
...     "_table": {
...         "cols": {"year": "$anchor_year - $anchor_age"},
...         "join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}},
...     },
...     "lab": {"code": "$test", "time": "$dischtime"},
... })
>>> sorted(tc.source_columns())
['anchor_age', 'anchor_year', 'patient_id', 'stay_id', 'test']

Source code in MEDS_extract/config.py

def source_columns(self) -> set[str]:
    """Columns that must be read from this table's source parquet file.

    Aggregates the subject_id source columns, the join left key, derived-column
    inputs, and all event-referenced columns — minus columns produced by
    ``_table.cols`` (derived) or pulled in by the join (come from elsewhere).

    Examples:
        >>> tc = TableConfig.parse("labs", {
        ...     "_defaults": {"subject_id": "$patient_id"},
        ...     "_table": {
        ...         "cols": {"year": "$anchor_year - $anchor_age"},
        ...         "join": {"stays": {"key": "stay_id", "cols": ["dischtime"]}},
        ...     },
        ...     "lab": {"code": "$test", "time": "$dischtime"},
        ... })
        >>> sorted(tc.source_columns())
        ['anchor_age', 'anchor_year', 'patient_id', 'stay_id', 'test']
    """
    cols: set[str] = set()
    if self.subject_id_node is not None:
        cols.update(self.subject_id_node.referenced_columns)
    else:
        cols.add("subject_id")

    for node in self.cols.values():
        cols.update(node.referenced_columns)

    if self.join is not None:
        cols.update(self.join.left_on)

    for event in self.events:
        cols.update(event.referenced_columns)

    return cols - self.col_outputs - self.joined_columns

source_files(dir)

Resolve the list of source files for this table under dir.

Delegates to :func:MEDS_extract.io.resolve_source_files. Returns a non-empty list; raises FileNotFoundError or ValueError (on layout ambiguity) otherwise.

Source code in MEDS_extract/config.py

def source_files(self, dir: Path | UPath) -> list[Path | UPath]:
    """Resolve the list of source files for this table under ``dir``.

    Delegates to :func:`MEDS_extract.io.resolve_source_files`. Returns a
    non-empty list; raises ``FileNotFoundError`` or ``ValueError`` (on
    layout ambiguity) otherwise.
    """
    return resolve_source_files(dir, self.input_prefix)

compile_metadata_block(block, component_cols, *, code_template_str, context='')

Compile and validate one _metadata prefix block against its declaring event.

A _metadata entry is a mapping of output column name → dftly expression, evaluated over the raw metadata table. The contract is name matching: produced columns whose names match the declaring event’s code-referenced component columns (component_cols) are the join keys; every other produced column is metadata output attached to the matched codes. Producing a strict subset of the component columns is a partial match — the metadata broadcasts to every code sharing the produced key values.

Values carry exactly dftly’s semantics — nothing is reinterpreted. In particular a bare, unquoted word is a string literal, not a column reference: description: label produces the constant text "label", while description: $label reads the raw label column.

Parameters:

Name	Type	Description	Default
block	Mapping[str, Any]	The raw per-prefix _metadata mapping.	required
component_cols	frozenset[str] | set[str]	Source columns referenced by the declaring event’s code expression.	required
code_template_str	str	The code expression string, for error messages.	required
context	str	Optional label naming the declaring event/prefix in error messages.	''

Raises:

Type	Description
ValueError	On every config mistake — a literal code (no components to match on), a non-mapping or empty block, a reserved output name, an unparsable expression, a block producing no join keys, or a block producing only join keys.

Examples:

>>> compiled = compile_metadata_block(
...     {"itemid": "$omop_source_code", "description": "$label", "vocab": '"MIMIC-IV"'},
...     {"itemid"},
...     code_template_str='f"CHART//{$itemid}"',
... )
>>> compiled.key_cols, compiled.output_cols
(('itemid',), ('description', 'vocab'))
>>> sorted(compiled.referenced_columns)
['label', 'omop_source_code']

Bare strings are dftly string literals, NOT column references — the block below produces the constant "label" for every row and references no metadata column at all (the 0.6.x shorthand must be migrated to $label):

>>> compiled = compile_metadata_block(
...     {"itemid": "$itemid", "description": "label"},
...     {"itemid"},
...     code_template_str='f"CHART//{$itemid}"',
... )
>>> compiled.exprs["description"]
Literal('label')
>>> sorted(compiled.referenced_columns)
['itemid']

parent_codes is an ordinary dftly output too — the multi-case matcher shape is a chained conditional, with the omitted final else yielding a real null for unmatched rows:

>>> compiled = compile_metadata_block(
...     {
...         "icd_code": "$icd_code",
...         "parent_codes": 'f"ICD{$icd_version}CM/{$icd_code}" if $icd_version == "9"',
...     },
...     {"icd_code"},
...     code_template_str="$icd_code",
... )
>>> type(compiled.exprs["parent_codes"]).__name__
'Conditional'

Producing a subset of the component columns is a partial match; producing none is an error naming the components the event offers:

>>> compiled = compile_metadata_block(
...     {"medication_name": "$medication_name", "description": "$drug_class"},
...     {"medication_name", "dose"},
...     code_template_str='f"{$medication_name}//{$dose}"',
... )
>>> compiled.key_cols
('medication_name',)
>>> compile_metadata_block(
...     {"description": "$label"},
...     {"itemid"},
...     code_template_str='f"CHART//{$itemid}"',
... )
Traceback (most recent call last):
    ...
ValueError: _metadata block produces no join-key columns: none of its produced column
names ['description'] match the code expression's component columns. At least one produced
column must be named after a component to serve as a join key. Component columns available
on this event: ['itemid'] (from code expression 'f"CHART//{$itemid}"').

A block with keys but no metadata outputs is also an error:

>>> compile_metadata_block(
...     {"itemid": "$itemid"}, {"itemid"}, code_template_str='f"CHART//{$itemid}"'
... )
Traceback (most recent call last):
    ...
ValueError: _metadata block produces only join-key columns ['itemid'] and no metadata
outputs. Add at least one output column (e.g. 'description') whose name does not match a
code component.

code/code_template are reserved output names — except that code may be a join key when the code expression references a source column literally named code (the ICD/OMOP vocabulary shape):

>>> compile_metadata_block(
...     {"code": "$icd_code", "description": "$label"},
...     {"icd_code"},
...     code_template_str="$icd_code",
... )
Traceback (most recent call last):
    ...
ValueError: _metadata output column name(s) ['code'] are reserved: 'code' and
'code_template' are generated by the pipeline and cannot be overwritten.
>>> compiled = compile_metadata_block(
...     {"code": "$code", "description": "$long_title"},
...     {"code"},
...     code_template_str='f"ICD//{$code}"',
... )
>>> compiled.key_cols
('code',)

Values that are not valid dftly fail naming the offending output column. A list (write a coalesce(...) for column fallback, or a conditional for a multi-case parent_codes), a bare {col} interpolation (a dftly f-string is f"LOINC/{$code}"), and a mapping all land here:

>>> compile_metadata_block(
...     {"itemid": "$itemid", "description": ["special_title", "title"]},
...     {"itemid"},
...     code_template_str='f"CHART//{$itemid}"',
... )
Traceback (most recent call last):
    ...
ValueError: _metadata column 'description' failed to parse as a dftly expression:
No matching node found for value: ['special_title', 'title'].
>>> compile_metadata_block(
...     {"icd_code": "$icd_code", "prefix": "LOINC/{icd_code}"},
...     {"icd_code"},
...     code_template_str="$icd_code",
... )
Traceback (most recent call last):
    ...
ValueError: _metadata column 'prefix' failed to parse as a dftly expression: ...
>>> compile_metadata_block(
...     {
...         "icd_code": "$icd_code",
...         "parent_codes": {"ICD{icd_version}CM/{icd_code}": {"icd_version": "9"}},
...     },
...     {"icd_code"},
...     code_template_str="$icd_code",
... )
Traceback (most recent call last):
    ...
ValueError: _metadata column 'parent_codes' failed to parse as a dftly expression: ...

Source code in MEDS_extract/config.py

def compile_metadata_block(
    block: Mapping[str, Any],
    component_cols: frozenset[str] | set[str],
    *,
    code_template_str: str,
    context: str = "",
) -> CompiledMetadataBlock:
    """Compile and validate one ``_metadata`` prefix block against its declaring event.

    A ``_metadata`` entry is a mapping of **output column name → dftly expression**,
    evaluated over the raw metadata table. The contract is *name matching*: produced
    columns whose names match the declaring event's code-referenced component columns
    (``component_cols``) are the **join keys**; every other produced column is metadata
    output attached to the matched codes. Producing a strict subset of the component
    columns is a partial match — the metadata broadcasts to every code sharing the
    produced key values.

    Values carry **exactly dftly's semantics** — nothing is reinterpreted. In
    particular a bare, unquoted word is a string *literal*, not a column reference:
    ``description: label`` produces the constant text ``"label"``, while
    ``description: $label`` reads the raw ``label`` column.

    Args:
        block: The raw per-prefix ``_metadata`` mapping.
        component_cols: Source columns referenced by the declaring event's ``code``
            expression.
        code_template_str: The code expression string, for error messages.
        context: Optional label naming the declaring event/prefix in error messages.

    Raises:
        ValueError: On every config mistake — a literal code (no components to match
            on), a non-mapping or empty block, a reserved output name, an unparsable
            expression, a block producing no join keys, or a block producing *only*
            join keys.

    Examples:
        >>> compiled = compile_metadata_block(
        ...     {"itemid": "$omop_source_code", "description": "$label", "vocab": '"MIMIC-IV"'},
        ...     {"itemid"},
        ...     code_template_str='f"CHART//{$itemid}"',
        ... )
        >>> compiled.key_cols, compiled.output_cols
        (('itemid',), ('description', 'vocab'))
        >>> sorted(compiled.referenced_columns)
        ['label', 'omop_source_code']

        Bare strings are dftly string literals, NOT column references — the block
        below produces the constant ``"label"`` for every row and references no
        metadata column at all (the 0.6.x shorthand must be migrated to ``$label``):

        >>> compiled = compile_metadata_block(
        ...     {"itemid": "$itemid", "description": "label"},
        ...     {"itemid"},
        ...     code_template_str='f"CHART//{$itemid}"',
        ... )
        >>> compiled.exprs["description"]
        Literal('label')
        >>> sorted(compiled.referenced_columns)
        ['itemid']

        ``parent_codes`` is an ordinary dftly output too — the multi-case matcher
        shape is a chained conditional, with the omitted final ``else`` yielding a
        real null for unmatched rows:

        >>> compiled = compile_metadata_block(
        ...     {
        ...         "icd_code": "$icd_code",
        ...         "parent_codes": 'f"ICD{$icd_version}CM/{$icd_code}" if $icd_version == "9"',
        ...     },
        ...     {"icd_code"},
        ...     code_template_str="$icd_code",
        ... )
        >>> type(compiled.exprs["parent_codes"]).__name__
        'Conditional'

        Producing a subset of the component columns is a partial match; producing
        none is an error naming the components the event offers:

        >>> compiled = compile_metadata_block(
        ...     {"medication_name": "$medication_name", "description": "$drug_class"},
        ...     {"medication_name", "dose"},
        ...     code_template_str='f"{$medication_name}//{$dose}"',
        ... )
        >>> compiled.key_cols
        ('medication_name',)
        >>> compile_metadata_block(
        ...     {"description": "$label"},
        ...     {"itemid"},
        ...     code_template_str='f"CHART//{$itemid}"',
        ... )
        Traceback (most recent call last):
            ...
        ValueError: _metadata block produces no join-key columns: none of its produced column
        names ['description'] match the code expression's component columns. At least one produced
        column must be named after a component to serve as a join key. Component columns available
        on this event: ['itemid'] (from code expression 'f"CHART//{$itemid}"').

        A block with keys but no metadata outputs is also an error:

        >>> compile_metadata_block(
        ...     {"itemid": "$itemid"}, {"itemid"}, code_template_str='f"CHART//{$itemid}"'
        ... )
        Traceback (most recent call last):
            ...
        ValueError: _metadata block produces only join-key columns ['itemid'] and no metadata
        outputs. Add at least one output column (e.g. 'description') whose name does not match a
        code component.

        ``code``/``code_template`` are reserved output names — except that ``code``
        may be a *join key* when the code expression references a source column
        literally named ``code`` (the ICD/OMOP vocabulary shape):

        >>> compile_metadata_block(
        ...     {"code": "$icd_code", "description": "$label"},
        ...     {"icd_code"},
        ...     code_template_str="$icd_code",
        ... )
        Traceback (most recent call last):
            ...
        ValueError: _metadata output column name(s) ['code'] are reserved: 'code' and
        'code_template' are generated by the pipeline and cannot be overwritten.
        >>> compiled = compile_metadata_block(
        ...     {"code": "$code", "description": "$long_title"},
        ...     {"code"},
        ...     code_template_str='f"ICD//{$code}"',
        ... )
        >>> compiled.key_cols
        ('code',)

        Values that are not valid dftly fail naming the offending output column. A list
        (write a ``coalesce(...)`` for column fallback, or a conditional for a
        multi-case ``parent_codes``), a bare ``{col}`` interpolation (a dftly f-string
        is ``f"LOINC/{$code}"``), and a mapping all land here:

        >>> compile_metadata_block(
        ...     {"itemid": "$itemid", "description": ["special_title", "title"]},
        ...     {"itemid"},
        ...     code_template_str='f"CHART//{$itemid}"',
        ... )
        Traceback (most recent call last):
            ...
        ValueError: _metadata column 'description' failed to parse as a dftly expression:
        No matching node found for value: ['special_title', 'title'].
        >>> compile_metadata_block(
        ...     {"icd_code": "$icd_code", "prefix": "LOINC/{icd_code}"},
        ...     {"icd_code"},
        ...     code_template_str="$icd_code",
        ... )
        Traceback (most recent call last):
            ...
        ValueError: _metadata column 'prefix' failed to parse as a dftly expression: ...
        >>> compile_metadata_block(
        ...     {
        ...         "icd_code": "$icd_code",
        ...         "parent_codes": {"ICD{icd_version}CM/{icd_code}": {"icd_version": "9"}},
        ...     },
        ...     {"icd_code"},
        ...     code_template_str="$icd_code",
        ... )
        Traceback (most recent call last):
            ...
        ValueError: _metadata column 'parent_codes' failed to parse as a dftly expression: ...
    """
    ctx = f" ({context})" if context else ""

    if not isinstance(block, Mapping) or isinstance(block, str):
        raise ValueError(
            f"_metadata entry{ctx} must be a mapping of output column name -> dftly expression, "
            f"got {type(block).__name__}: {block!r}."
        )
    if not block:
        raise ValueError(
            f"_metadata entry{ctx} is empty. Each metadata prefix must map output column names "
            f"to dftly expressions over the raw metadata table."
        )
    component_cols = frozenset(component_cols)
    if not component_cols:
        raise ValueError(
            f"The code expression {code_template_str!r} is a literal: it references no source "
            "columns, so a literal code has no components to match metadata on. Add static "
            "metadata for literal codes to a pre-existing codes.parquet instead of a _metadata "
            "block."
        )

    # ``code`` name-matching a component is a join key (consumed by the component join,
    # never emitted as an output), so only non-component ``code`` and any
    # ``code_template`` are reserved-name violations.
    reserved = sorted(
        k for k in block if k in METADATA_RESERVED_COLS and not (k == "code" and k in component_cols)
    )
    if reserved:
        raise ValueError(
            f"_metadata output column name(s) {reserved} are reserved: 'code' and "
            "'code_template' are generated by the pipeline and cannot be overwritten."
        )

    parser = Parser()
    exprs: dict[str, NodeBase] = {}
    for out_col, raw_expr in block.items():
        try:
            exprs[out_col] = raw_expr if isinstance(raw_expr, NodeBase) else parser(raw_expr)
        except Exception as e:
            raise ValueError(
                f"_metadata column {out_col!r}{ctx} failed to parse as a dftly expression: {e}"
            ) from e

    key_cols = tuple(sorted(set(exprs) & component_cols))
    if not key_cols:
        raise ValueError(
            f"_metadata block{ctx} produces no join-key columns: none of its produced column "
            f"names {sorted(block)} match the code expression's component columns. At least one "
            f"produced column must be named after a component to serve as a join key. Component "
            f"columns available on this event: {sorted(component_cols)} (from code expression "
            f"{code_template_str!r})."
        )

    output_cols = tuple(k for k in block if k not in key_cols)
    if not output_cols:
        raise ValueError(
            f"_metadata block{ctx} produces only join-key columns {sorted(key_cols)} and no "
            f"metadata outputs. Add at least one output column (e.g. 'description') whose name "
            f"does not match a code component."
        )

    return CompiledMetadataBlock(
        exprs=exprs, key_cols=key_cols, output_cols=output_cols, code_template=code_template_str
    )

compile_self_metadata_block(block, component_cols, *, code_template_str, context='')

Compile and validate one _self metadata block against its declaring event.

A _self block maps output column name → dftly expression, evaluated over the event’s own prepared source frame (post-join, post-_table.cols) during event extraction — never over a re-scanned raw table. Join keys are implicit: every code component pairs with the outputs row-wise exactly, so unlike an external :func:compile_metadata_block entry, the block never restates key columns — key_cols is always the full component set, and every produced column is a metadata output.

Raises:

Type	Description
ValueError	On a literal code (no components to attach through), a non-mapping or empty block, a reserved output name, an output name colliding with a component column (keys are implicit — the name is already taken), or an unparsable expression.

Examples:

>>> compiled = compile_self_metadata_block(
...     {"description": "$long_label", "vocab": '"LOCAL"'},
...     {"itemid"},
...     code_template_str='f"CHART//{$itemid}"',
... )
>>> compiled.key_cols, compiled.output_cols
(('itemid',), ('description', 'vocab'))
>>> sorted(compiled.referenced_columns)
['long_label']

Output names may not collide with the code’s components (keys are implicit):

>>> compile_self_metadata_block(
...     {"itemid": "$other"}, {"itemid"}, code_template_str='f"CHART//{$itemid}"'
... )
Traceback (most recent call last):
    ...
ValueError: _self metadata output column name(s) ['itemid'] collide with the code
expression's component columns. _self join keys are implicit (every component pairs
row-wise); name the outputs differently.

A literal code has no components for the reducer to attach through:

>>> compile_self_metadata_block(
...     {"description": "$label"}, set(), code_template_str='"ADMISSION"'
... )
Traceback (most recent call last):
    ...
ValueError: The code expression '"ADMISSION"' is a literal: it references no source
columns, so a literal code has no components to match metadata on. Add static metadata
for literal codes to a pre-existing codes.parquet instead of a _metadata block.

Source code in MEDS_extract/config.py

def compile_self_metadata_block(
    block: Mapping[str, Any],
    component_cols: frozenset[str] | set[str],
    *,
    code_template_str: str,
    context: str = "",
) -> CompiledMetadataBlock:
    """Compile and validate one ``_self`` metadata block against its declaring event.

    A ``_self`` block maps **output column name → dftly expression**, evaluated over the
    event's own prepared source frame (post-join, post-``_table.cols``) during event
    extraction — never over a re-scanned raw table. Join keys are implicit: every code
    component pairs with the outputs row-wise exactly, so unlike an external
    :func:`compile_metadata_block` entry, the block never restates key columns —
    ``key_cols`` is always the full component set, and every produced column is a
    metadata output.

    Raises:
        ValueError: On a literal code (no components to attach through), a non-mapping
            or empty block, a reserved output name, an output name colliding with a
            component column (keys are implicit — the name is already taken), or an
            unparsable expression.

    Examples:
        >>> compiled = compile_self_metadata_block(
        ...     {"description": "$long_label", "vocab": '"LOCAL"'},
        ...     {"itemid"},
        ...     code_template_str='f"CHART//{$itemid}"',
        ... )
        >>> compiled.key_cols, compiled.output_cols
        (('itemid',), ('description', 'vocab'))
        >>> sorted(compiled.referenced_columns)
        ['long_label']

        Output names may not collide with the code's components (keys are implicit):

        >>> compile_self_metadata_block(
        ...     {"itemid": "$other"}, {"itemid"}, code_template_str='f"CHART//{$itemid}"'
        ... )
        Traceback (most recent call last):
            ...
        ValueError: _self metadata output column name(s) ['itemid'] collide with the code
        expression's component columns. _self join keys are implicit (every component pairs
        row-wise); name the outputs differently.

        A literal code has no components for the reducer to attach through:

        >>> compile_self_metadata_block(
        ...     {"description": "$label"}, set(), code_template_str='"ADMISSION"'
        ... )
        Traceback (most recent call last):
            ...
        ValueError: The code expression '"ADMISSION"' is a literal: it references no source
        columns, so a literal code has no components to match metadata on. Add static metadata
        for literal codes to a pre-existing codes.parquet instead of a _metadata block.
    """
    ctx = f" ({context})" if context else ""

    if not isinstance(block, Mapping) or isinstance(block, str):
        raise ValueError(
            f"_self metadata entry{ctx} must be a mapping of output column name -> dftly "
            f"expression, got {type(block).__name__}: {block!r}."
        )
    if not block:
        raise ValueError(
            f"_self metadata entry{ctx} is empty. Map output column names to dftly expressions "
            f"over the event's own source frame."
        )
    component_cols = frozenset(component_cols)
    if not component_cols:
        raise ValueError(
            f"The code expression {code_template_str!r} is a literal: it references no source "
            "columns, so a literal code has no components to match metadata on. Add static "
            "metadata for literal codes to a pre-existing codes.parquet instead of a _metadata "
            "block."
        )
    reserved = sorted(k for k in block if k in METADATA_RESERVED_COLS)
    if reserved:
        raise ValueError(
            f"_self metadata output column name(s) {reserved}{ctx} are reserved: 'code' and "
            "'code_template' are generated by the pipeline and cannot be overwritten."
        )
    colliding = sorted(set(block) & component_cols)
    if colliding:
        raise ValueError(
            f"_self metadata output column name(s) {colliding}{ctx} collide with the code "
            f"expression's component columns. _self join keys are implicit (every component "
            f"pairs row-wise); name the outputs differently."
        )

    parser = Parser()
    exprs: dict[str, NodeBase] = {}
    for out_col, raw_expr in block.items():
        try:
            exprs[out_col] = raw_expr if isinstance(raw_expr, NodeBase) else parser(raw_expr)
        except Exception as e:
            raise ValueError(
                f"_self metadata column {out_col!r}{ctx} failed to parse as a dftly expression: {e}"
            ) from e

    return CompiledMetadataBlock(
        exprs=exprs,
        key_cols=tuple(sorted(component_cols)),
        output_cols=tuple(block),
        code_template=code_template_str,
    )

resolve_config_path(path)

Resolve a config-file reference that may use pkg:// syntax.

The one shared helper behind every place a MESSY reference is consumed — MessyConfig.load (so every stage’s MESSY_config_fp accepts pkg://) and thereby every MessyConfig.load consumer — via MEDS-transforms’ resolve_pkg_path, the same syntax MEDS_transform-pipeline accepts for pipeline configs.

Examples:

>>> resolve_config_path("pkg://MEDS_extract.configs._extract.yaml").name
'_extract.yaml'
>>> resolve_config_path("/some/local/messy.yaml")
PosixPath('/some/local/messy.yaml')

Source code in MEDS_extract/config.py

def resolve_config_path(path: str | Path) -> Path:
    """Resolve a config-file reference that may use ``pkg://`` syntax.

    The one shared helper behind every place a MESSY reference is consumed —
    ``MessyConfig.load`` (so every stage's ``MESSY_config_fp`` accepts
    ``pkg://``) and thereby every ``MessyConfig.load`` consumer — via
    MEDS-transforms' ``resolve_pkg_path``, the same syntax ``MEDS_transform-pipeline``
    accepts for pipeline configs.

    Examples:
        >>> resolve_config_path("pkg://MEDS_extract.configs._extract.yaml").name
        '_extract.yaml'
        >>> resolve_config_path("/some/local/messy.yaml")
        PosixPath('/some/local/messy.yaml')
    """
    if str(path).startswith(PKG_PFX):
        return Path(str(resolve_pkg_path(str(path))))
    return Path(path)

user_local_path(p, *, field='path')

Normalize a user-supplied local path; reject URLs with an actionable error.

Both CLIs run with hydra.job.chdir=false, so a relative path means exactly what the invoking shell suggests; expanduser/resolve normalize ~, ., and symlinks into the unambiguous absolute form the child processes need. URL-shaped values are rejected rather than silently collapsed into a local relative path (Path("s3://b/x") becomes the literal directory s3:/b/x under CWD — see #213): these fields are local-only by design, and the download layer is the one remote-data ingress.

Examples:

>>> user_local_path("/already/absolute")
PosixPath('/already/absolute')
>>> user_local_path("relative_dir").is_absolute()
True
>>> user_local_path("s3://bucket/raw", field="output_dir")
Traceback (most recent call last):
    ...
ValueError: output_dir='s3://bucket/raw' is a URL, but output_dir must be a local
filesystem path. Remote data enters through the download layer (e.g. a `type: fsspec`
source in the spec's `sources:` block): fetch to local disk first, then point
output_dir at the local copy.

Source code in MEDS_extract/config.py

def user_local_path(p: str | Path, *, field: str = "path") -> Path:
    """Normalize a user-supplied local path; reject URLs with an actionable error.

    Both CLIs run with ``hydra.job.chdir=false``, so a relative path means exactly
    what the invoking shell suggests; ``expanduser``/``resolve`` normalize ``~``,
    ``.``, and symlinks into the unambiguous absolute form the child processes need.
    URL-shaped values are rejected rather than silently collapsed into a local
    relative path (``Path("s3://b/x")`` becomes the literal directory ``s3:/b/x``
    under CWD — see #213): these fields are local-only by design, and the download
    layer is the one remote-data ingress.

    Examples:
        >>> user_local_path("/already/absolute")
        PosixPath('/already/absolute')
        >>> user_local_path("relative_dir").is_absolute()
        True
        >>> user_local_path("s3://bucket/raw", field="output_dir")
        Traceback (most recent call last):
            ...
        ValueError: output_dir='s3://bucket/raw' is a URL, but output_dir must be a local
        filesystem path. Remote data enters through the download layer (e.g. a `type: fsspec`
        source in the spec's `sources:` block): fetch to local disk first, then point
        output_dir at the local copy.
    """
    s = str(p)
    if "://" in s:
        raise ValueError(
            f"{field}={s!r} is a URL, but {field} must be a local filesystem path. "
            "Remote data enters through the download layer (e.g. a `type: fsspec` source "
            f"in the spec's `sources:` block): fetch to local disk first, then point "
            f"{field} at the local copy."
        )
    return Path(s).expanduser().resolve()