Skip to content

DatabaseSampler

Working with a full patient database during development slows feedback cycles. DatabaseSampler solves this by filtering every domain table down to a reproducible subset of patients — defined by fraction and seed — before any PhenEx phenotypes are evaluated. All filtering is expressed as SQL and runs inside the database.

How it works

Each patient is assigned to a stable group using:

abs(hash(str(PERSON_ID) || str(seed))) % denom == 0

where denom = round(1 / fraction). The fraction controls how large the group is; the seed controls which specific patients end up in it.

Reproducible. The same fraction and seed always return exactly the same patients — across runs, across machines, and across team members. There is no random state: the result is fully determined by the two parameters.

Stable. Because selection is based on a hash of the ID, adding new patients to the database does not change which existing patients are selected. Your development sample stays consistent as the underlying data grows.

Type-agnostic. PERSON_ID can be an integer, a UUID, or any string — the algorithm works identically for all types, with no special-casing.

Runs inside the database. The filter is evaluated as a SQL expression. No data is transferred to Python until you explicitly request it.

Quick start

from phenex.util import DatabaseSampler

# 10 % of patients, default seed
sampler = DatabaseSampler(fraction=0.1)
sampled_tables = sampler.sample(mapped_tables)   # lazy — no data moves yet

# sampled_tables is a dict with the same keys as mapped_tables;
# pass it to a Cohort or use it directly.
print(sampler.describe())

sample() returns a new mapped_tables dict. The original dict is not modified. Every domain that has a PERSON_ID column is filtered to the sampled patients; domains without PERSON_ID are passed through unchanged.

Using the sampler in a Study

Calling sample() by hand is useful for exploration, but the usual pattern is to attach the sampler to the Study's Database.

from phenex import Cohort, Database, Study
from phenex.util import DatabaseSampler
from phenex.mappers import OMOPDomains

study_db = Database(
    mapper=OMOPDomains,
    connector=con,                                   # con: your connector
    sampler=DatabaseSampler(fraction=0.1, seed=42),  # 10% reproducible sample
)

cohort_a = Cohort(name="cohort_a", entry_criterion=entry_a, inclusions=inclusions, exclusions=exclusions)
cohort_b = Cohort(name="cohort_b", entry_criterion=entry_b, inclusions=inclusions, exclusions=exclusions)

study = Study(path="./results", name="my_study_sampled", cohorts=[cohort_a, cohort_b], database=study_db)
study.execute(overwrite=True, lazy_execution=True)

Inspecting the sample

fetch_person_ids() executes one database round-trip and loads the sampled IDs into Python:

sampler = DatabaseSampler(fraction=0.1, seed=42)
sampler.sample(mapped_tables)

ids = sampler.fetch_person_ids()   # sorted list
print(sampler.person_id_count)     # e.g. 1 042
print(sampler.person_ids[:5])      # first five IDs

describe() prints a human-readable configuration summary and is safe to call at any point:

DatabaseSampler
  fraction   : 0.1
  seed       : 42
  denom      : 10  (10 equal groups)
  filter     : abs(hash(str(PERSON_ID) || '42')) % 10 = 0
  sampled    : yes -- call fetch_person_ids() to inspect
  patients   : 1,042
  first 10   : [3, 17, 28, ...]

API reference

Filter every domain table to a reproducible subset of patients.

Each patient is assigned to a reproducible group by hashing their PERSON_ID together with the seed. The fraction controls how large that group is; the seed controls which specific patients end up in it. The same fraction and seed always return the same patients.

Parameters:

Name Type Description Default
fraction float | None

Proportion of patients to include, in [0.0, 1.0]. fraction=0.0 always returns an empty sample.

 None
seed int

Any integer. Controls which patients are selected. Default 42. Different seeds produce different cohorts of the same size.

 42

Attributes:

Name Type Description
person_ids list[Any] | None

Sorted list of sampled PERSON_IDs. Set after fetch_person_ids().
person_id_count int | None

Number of sampled patients. Set after fetch_person_ids().
Example

sampler = DatabaseSampler(fraction=0.1, seed=42) sampled = sampler.sample(mapped_tables) # lazy, no data moved

print(sampler.describe())

ids = sampler.fetch_person_ids() # one database round-trip print(sampler.person_id_count)

Different cohort, same size

sampler2 = DatabaseSampler(fraction=0.1, seed=99)

Source code in phenex/util/database_sampler.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
class DatabaseSampler:
    """Filter every domain table to a reproducible subset of patients.

    Each patient is assigned to a reproducible group by hashing their PERSON_ID
    together with the seed. The fraction controls how large that group is; the
    seed controls which specific patients end up in it. The same fraction and
    seed always return the same patients.

    Args:
        fraction: Proportion of patients to include, in [0.0, 1.0].
                  fraction=0.0 always returns an empty sample.
        seed:     Any integer. Controls which patients are selected. Default 42.
                  Different seeds produce different cohorts of the same size.

    Attributes:
        person_ids: Sorted list of sampled PERSON_IDs. Set after fetch_person_ids().
        person_id_count: Number of sampled patients. Set after fetch_person_ids().

    Example:
        sampler = DatabaseSampler(fraction=0.1, seed=42)
        sampled = sampler.sample(mapped_tables)   # lazy, no data moved

        print(sampler.describe())

        ids = sampler.fetch_person_ids()          # one database round-trip
        print(sampler.person_id_count)

        # Different cohort, same size
        sampler2 = DatabaseSampler(fraction=0.1, seed=99)
    """

    def __init__(self, fraction: float | None = None, seed: int = 42) -> None:
        if fraction is None:
            raise ValueError("fraction is required.")
        if isinstance(fraction, bool) or not isinstance(fraction, (int, float)):
            raise ValueError(
                f"fraction must be a number, got {type(fraction).__name__}."
            )
        if not (0.0 <= fraction <= 1.0):
            raise ValueError(f"fraction must be in [0.0, 1.0], got {fraction}.")
        if isinstance(seed, bool) or not isinstance(seed, int):
            raise ValueError(f"seed must be an integer, got {type(seed).__name__}.")

        self.fraction: float = fraction
        self.seed: int = seed
        self.denom: int = 0 if fraction == 0.0 else max(1, round(1.0 / fraction))

        self._person_ids_expr: ir.Table | None = None
        self.person_ids: list[Any] | None = None
        self.person_id_count: int | None = None

    def sample(self, mapped_tables: dict[str, Any]) -> dict[str, Any]:
        """Return a new mapped_tables dict filtered to the sampled patients.

        Builds SQL expressions only, nothing runs in the database yet.
        Domains that are None or have no PERSON_ID column are passed through
        unchanged.

        Args:
            mapped_tables: Dict[str, PhenexTable | None] from
                DomainsDictionary.get_mapped_tables().

        Returns:
            Dict[str, PhenexTable | None] with the same keys.

        Raises:
            KeyError: if mapped_tables does not contain a non-None "PERSON" entry.
            ValueError: if the PERSON table has no "PERSON_ID" column.
        """
        if "PERSON" not in mapped_tables or mapped_tables["PERSON"] is None:
            raise KeyError(
                "mapped_tables must contain a non-None 'PERSON' entry. "
                "Ensure mapped_tables includes PERSON."
            )

        person_table = mapped_tables["PERSON"].table
        if "PERSON_ID" not in person_table.columns:
            raise ValueError(
                "The PERSON table must have a 'PERSON_ID' column, but none was found. "
                f"Columns present: {list(person_table.columns)}"
            )
        self._person_ids_expr = self._sampled_person_ids(person_table)
        self.person_ids = None
        self.person_id_count = None

        if self.fraction == 1.0:
            return dict(mapped_tables)

        result: dict[str, Any] = {}
        for domain_name, domain in mapped_tables.items():
            if domain is None:
                result[domain_name] = None
                continue
            if "PERSON_ID" not in domain.table.columns:
                result[domain_name] = domain
                continue

            filtered_ibis = domain.table.join(
                self._person_ids_expr, "PERSON_ID"
            ).select(domain.table.columns)
            result[domain_name] = type(domain)(
                filtered_ibis,
                name=domain.NAME_TABLE,
                column_mapping=domain.column_mapping,
            )

        return result

    def __repr__(self) -> str:
        """Concise representation showing the config that defines the sample."""
        return f"{self.__class__.__name__}(fraction={self.fraction}, seed={self.seed})"

    def to_dict(self) -> dict:
        """Serialize to a JSON-safe dict for cohort snapshot storage."""
        return {
            "class_name": self.__class__.__name__,
            "fraction": self.fraction,
            "seed": self.seed,
        }

    def fetch_person_ids(self) -> list[Any]:
        """Fetch sampled PERSON_IDs from the database into a sorted Python list.

        This is the only method that moves data to Python. Call it when you
        need the ID list for inspection, logging, or an external handoff.

        Populates self.person_ids and self.person_id_count.
        Can be called multiple times, re-fetches each time.

        Returns:
            list of PERSON_ID values, sorted ascending.

        Raises:
            RuntimeError: if called before .sample().
        """
        if self._person_ids_expr is None:
            raise RuntimeError("Call .sample(mapped_tables) before fetch_person_ids().")
        df = self._person_ids_expr.order_by("PERSON_ID").execute()
        self.person_ids = df["PERSON_ID"].tolist()
        self.person_id_count = len(self.person_ids)
        return self.person_ids

    def describe(self) -> str:
        """Return a plain-text summary of this sampler's configuration.

        Safe to call at any lifecycle stage. Patient count appears only after
        fetch_person_ids() has been called.

        Returns:
            str: Human-readable configuration summary.
        """
        if self.fraction == 0.0:
            denom_filter = [
                "  denom      : 0  (fraction=0.0, no patients selected)",
                "  filter     : fraction=0.0 -> always empty",
            ]
        elif self.fraction == 1.0:
            denom_filter = [
                "  denom      : 1  (fraction=1.0, all patients selected)",
                "  filter     : fraction=1.0 -> keep all (hash skipped)",
            ]
        else:
            denom_filter = [
                f"  denom      : {self.denom}  ({self.denom} equal groups)",
                f"  filter     : abs(hash(str(PERSON_ID) || '{self.seed}')) % {self.denom} = 0",
            ]

        lines = [
            "DatabaseSampler",
            f"  fraction   : {self.fraction}",
            f"  seed       : {self.seed}",
            *denom_filter,
            f"  sampled    : {'yes -- call fetch_person_ids() to inspect' if self._person_ids_expr is not None else 'no -- call .sample() first'}",
        ]
        if self.person_id_count is not None:
            lines.append(f"  patients   : {self.person_id_count:,}")
            lines.append(f"  first 10   : {self.person_ids[:10]}")
        else:
            lines.append("  patients   : (call .fetch_person_ids() to load)")

        return "\n".join(lines)

    def _sampled_person_ids(self, person_table: ir.Table) -> ir.Table:
        """Return a lazy expression of sampled PERSON_IDs via hash filter.

        The filter is:
            abs(hash(str(PERSON_ID) || str(seed))) % denom == 0

        Casting PERSON_ID to string before concatenating the seed makes the
        algorithm type-agnostic: integer IDs, UUID strings, and any VARCHAR
        type all work without branching. Every seed produces a completely
        independent scrambling of patients; different seeds always select
        different cohorts of the same size.

        Returns:
            ir.Table: Lazy expression of distinct sampled PERSON_IDs.
        """
        if self.fraction == 0.0:
            return person_table.limit(0).select("PERSON_ID").distinct()

        if self.fraction == 1.0:
            return person_table.select("PERSON_ID").distinct()

        return (
            person_table.filter(
                person_table["PERSON_ID"]
                .cast("string")
                .concat(str(self.seed))
                .hash()
                .abs()
                % self.denom
                == 0
            )
            .select("PERSON_ID")
            .distinct()
        )

__repr__()

Concise representation showing the config that defines the sample.

Source code in phenex/util/database_sampler.py 
116
117
118
def __repr__(self) -> str:
    """Concise representation showing the config that defines the sample."""
    return f"{self.__class__.__name__}(fraction={self.fraction}, seed={self.seed})"

describe()

Return a plain-text summary of this sampler's configuration.

Safe to call at any lifecycle stage. Patient count appears only after fetch_person_ids() has been called.

Returns:

Name Type Description
str str

Human-readable configuration summary.
Source code in phenex/util/database_sampler.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def describe(self) -> str:
    """Return a plain-text summary of this sampler's configuration.

    Safe to call at any lifecycle stage. Patient count appears only after
    fetch_person_ids() has been called.

    Returns:
        str: Human-readable configuration summary.
    """
    if self.fraction == 0.0:
        denom_filter = [
            "  denom      : 0  (fraction=0.0, no patients selected)",
            "  filter     : fraction=0.0 -> always empty",
        ]
    elif self.fraction == 1.0:
        denom_filter = [
            "  denom      : 1  (fraction=1.0, all patients selected)",
            "  filter     : fraction=1.0 -> keep all (hash skipped)",
        ]
    else:
        denom_filter = [
            f"  denom      : {self.denom}  ({self.denom} equal groups)",
            f"  filter     : abs(hash(str(PERSON_ID) || '{self.seed}')) % {self.denom} = 0",
        ]

    lines = [
        "DatabaseSampler",
        f"  fraction   : {self.fraction}",
        f"  seed       : {self.seed}",
        *denom_filter,
        f"  sampled    : {'yes -- call fetch_person_ids() to inspect' if self._person_ids_expr is not None else 'no -- call .sample() first'}",
    ]
    if self.person_id_count is not None:
        lines.append(f"  patients   : {self.person_id_count:,}")
        lines.append(f"  first 10   : {self.person_ids[:10]}")
    else:
        lines.append("  patients   : (call .fetch_person_ids() to load)")

    return "\n".join(lines)

fetch_person_ids()

Fetch sampled PERSON_IDs from the database into a sorted Python list.

This is the only method that moves data to Python. Call it when you need the ID list for inspection, logging, or an external handoff.

Populates self.person_ids and self.person_id_count. Can be called multiple times, re-fetches each time.

Returns:

Type Description
list[Any]

list of PERSON_ID values, sorted ascending.

Raises:

Type Description
RuntimeError

if called before .sample().
Source code in phenex/util/database_sampler.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def fetch_person_ids(self) -> list[Any]:
    """Fetch sampled PERSON_IDs from the database into a sorted Python list.

    This is the only method that moves data to Python. Call it when you
    need the ID list for inspection, logging, or an external handoff.

    Populates self.person_ids and self.person_id_count.
    Can be called multiple times, re-fetches each time.

    Returns:
        list of PERSON_ID values, sorted ascending.

    Raises:
        RuntimeError: if called before .sample().
    """
    if self._person_ids_expr is None:
        raise RuntimeError("Call .sample(mapped_tables) before fetch_person_ids().")
    df = self._person_ids_expr.order_by("PERSON_ID").execute()
    self.person_ids = df["PERSON_ID"].tolist()
    self.person_id_count = len(self.person_ids)
    return self.person_ids

sample(mapped_tables)

Return a new mapped_tables dict filtered to the sampled patients.

Builds SQL expressions only, nothing runs in the database yet. Domains that are None or have no PERSON_ID column are passed through unchanged.

Parameters:

Name Type Description Default
mapped_tables dict[str, Any]

Dict[str, PhenexTable | None] from DomainsDictionary.get_mapped_tables().

 required

Returns:

Type Description
dict[str, Any]

Dict[str, PhenexTable | None] with the same keys.

Raises:

Type Description
KeyError

if mapped_tables does not contain a non-None "PERSON" entry.
ValueError

if the PERSON table has no "PERSON_ID" column.
Source code in phenex/util/database_sampler.py 
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def sample(self, mapped_tables: dict[str, Any]) -> dict[str, Any]:
    """Return a new mapped_tables dict filtered to the sampled patients.

    Builds SQL expressions only, nothing runs in the database yet.
    Domains that are None or have no PERSON_ID column are passed through
    unchanged.

    Args:
        mapped_tables: Dict[str, PhenexTable | None] from
            DomainsDictionary.get_mapped_tables().

    Returns:
        Dict[str, PhenexTable | None] with the same keys.

    Raises:
        KeyError: if mapped_tables does not contain a non-None "PERSON" entry.
        ValueError: if the PERSON table has no "PERSON_ID" column.
    """
    if "PERSON" not in mapped_tables or mapped_tables["PERSON"] is None:
        raise KeyError(
            "mapped_tables must contain a non-None 'PERSON' entry. "
            "Ensure mapped_tables includes PERSON."
        )

    person_table = mapped_tables["PERSON"].table
    if "PERSON_ID" not in person_table.columns:
        raise ValueError(
            "The PERSON table must have a 'PERSON_ID' column, but none was found. "
            f"Columns present: {list(person_table.columns)}"
        )
    self._person_ids_expr = self._sampled_person_ids(person_table)
    self.person_ids = None
    self.person_id_count = None

    if self.fraction == 1.0:
        return dict(mapped_tables)

    result: dict[str, Any] = {}
    for domain_name, domain in mapped_tables.items():
        if domain is None:
            result[domain_name] = None
            continue
        if "PERSON_ID" not in domain.table.columns:
            result[domain_name] = domain
            continue

        filtered_ibis = domain.table.join(
            self._person_ids_expr, "PERSON_ID"
        ).select(domain.table.columns)
        result[domain_name] = type(domain)(
            filtered_ibis,
            name=domain.NAME_TABLE,
            column_mapping=domain.column_mapping,
        )

    return result

to_dict()

Serialize to a JSON-safe dict for cohort snapshot storage.

Source code in phenex/util/database_sampler.py 
120
121
122
123
124
125
126
def to_dict(self) -> dict:
    """Serialize to a JSON-safe dict for cohort snapshot storage."""
    return {
        "class_name": self.__class__.__name__,
        "fraction": self.fraction,
        "seed": self.seed,
    }