climate_ref.results.resources

Read surface for per-execution resource measurements.

ResourcesReader is reached via Reader.resources. It reads the resource columns on Execution and aggregates them into a ResourceProfile per diagnostic or per provider, which is the number a maintainer needs when sizing a worker.

Not all executions have resource data.

Confidence = Literal['good', 'thin', 'none'] module-attribute

How much weight a profile's recommendation can carry.

GroupBy = Literal['diagnostic', 'provider'] module-attribute

Axis a profile is aggregated over.

ResourceFilter

Declarative filter over the executions considered for a resource profile.

Every field is optional, and None means "do not constrain on this axis". diagnostic_contains/provider_contains are case-insensitive substring matches (OR-combined within each field), matching the semantics used by ExecutionGroupFilter.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

@attrs.frozen(kw_only=True)
class ResourceFilter:
    """
    Declarative filter over the executions considered for a resource profile.

    Every field is optional, and ``None`` means "do not constrain on this axis".
    ``diagnostic_contains``/``provider_contains`` are case-insensitive substring matches
    (OR-combined within each field),
    matching the semantics used by
    [ExecutionGroupFilter][climate_ref.results.executions.ExecutionGroupFilter].
    """

    diagnostic_contains: tuple[str, ...] | None = attrs.field(default=None, converter=_as_str_tuple)
    """Case-insensitive substring matches on diagnostic slug (OR-combined)."""

    provider_contains: tuple[str, ...] | None = attrs.field(default=None, converter=_as_str_tuple)
    """Case-insensitive substring matches on provider slug (OR-combined)."""

    since: datetime.datetime | None = None
    """Keep only executions created at or after this naive UTC timestamp."""

diagnostic_contains = attrs.field(default=None, converter=_as_str_tuple) class-attribute instance-attribute

Case-insensitive substring matches on diagnostic slug (OR-combined).

provider_contains = attrs.field(default=None, converter=_as_str_tuple) class-attribute instance-attribute

Case-insensitive substring matches on provider slug (OR-combined).

since = None class-attribute instance-attribute

Keep only executions created at or after this naive UTC timestamp.

ResourceMeasurementCollection

An immutable page of resource measurements plus collection-level metadata.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

@attrs.frozen(kw_only=True)
class ResourceMeasurementCollection:
    """An immutable page of resource measurements plus collection-level metadata."""

    items: tuple[ResourceMeasurementView, ...]
    """The measurements on this page."""

    total_count: int
    """Total measurements matching the filter before ``offset``/``limit``."""

    offset: int
    """Rows skipped before this page."""

    limit: int | None
    """Page size requested, or ``None`` when the whole result was returned."""

    def __iter__(self) -> Iterator[ResourceMeasurementView]:
        return iter(self.items)

    def __len__(self) -> int:
        return len(self.items)

    def to_pandas(self) -> pd.DataFrame:
        """
        DataFrame with one row per measurement.

        Columns are emitted explicitly even when the collection is empty,
        so callers can select columns without special-casing.
        """
        columns = [
            "execution_id",
            "provider",
            "diagnostic",
            "successful",
            "wall_seconds",
            "cpu_seconds",
            "peak_memory_bytes",
            "memory_source",
            "memory_limit_bytes",
            "cpu_limit",
            "resources_exclusive",
            "queue_seconds",
            "created_at",
        ]
        records = [
            {
                "execution_id": item.execution_id,
                "provider": item.provider_slug,
                "diagnostic": item.diagnostic_slug,
                "successful": item.successful,
                "wall_seconds": item.wall_seconds,
                "cpu_seconds": item.cpu_seconds,
                "peak_memory_bytes": item.peak_memory_bytes,
                "memory_source": item.memory_source,
                "memory_limit_bytes": item.memory_limit_bytes,
                "cpu_limit": item.cpu_limit,
                "resources_exclusive": item.resources_exclusive,
                "queue_seconds": item.queue_seconds,
                "created_at": item.created_at,
            }
            for item in self.items
        ]
        return pd.DataFrame.from_records(records, columns=columns)

items instance-attribute

The measurements on this page.

limit instance-attribute

Page size requested, or None when the whole result was returned.

offset instance-attribute

Rows skipped before this page.

total_count instance-attribute

Total measurements matching the filter before offset/limit.

to_pandas()

DataFrame with one row per measurement.

Columns are emitted explicitly even when the collection is empty, so callers can select columns without special-casing.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

def to_pandas(self) -> pd.DataFrame:
    """
    DataFrame with one row per measurement.

    Columns are emitted explicitly even when the collection is empty,
    so callers can select columns without special-casing.
    """
    columns = [
        "execution_id",
        "provider",
        "diagnostic",
        "successful",
        "wall_seconds",
        "cpu_seconds",
        "peak_memory_bytes",
        "memory_source",
        "memory_limit_bytes",
        "cpu_limit",
        "resources_exclusive",
        "queue_seconds",
        "created_at",
    ]
    records = [
        {
            "execution_id": item.execution_id,
            "provider": item.provider_slug,
            "diagnostic": item.diagnostic_slug,
            "successful": item.successful,
            "wall_seconds": item.wall_seconds,
            "cpu_seconds": item.cpu_seconds,
            "peak_memory_bytes": item.peak_memory_bytes,
            "memory_source": item.memory_source,
            "memory_limit_bytes": item.memory_limit_bytes,
            "cpu_limit": item.cpu_limit,
            "resources_exclusive": item.resources_exclusive,
            "queue_seconds": item.queue_seconds,
            "created_at": item.created_at,
        }
        for item in self.items
    ]
    return pd.DataFrame.from_records(records, columns=columns)

ResourceMeasurementView

One execution's resource measurement, detached from the ORM.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

@attrs.frozen(kw_only=True)
class ResourceMeasurementView:
    """One execution's resource measurement, detached from the ORM."""

    execution_id: int
    """Primary key of the underlying ``Execution`` row."""

    provider_slug: str
    """Owning provider's slug."""

    diagnostic_slug: str
    """Owning diagnostic's slug."""

    successful: bool | None
    """``True``/``False`` once the execution has finished, ``None`` while still running."""

    wall_seconds: float | None
    """Wall clock time taken by the execution, in seconds."""

    cpu_seconds: float | None
    """CPU time consumed by the execution and its children, in seconds."""

    peak_memory_bytes: int | None
    """Peak resident memory observed during the execution, in bytes."""

    memory_source: MemorySource | None
    """Provenance of ``peak_memory_bytes``."""

    memory_limit_bytes: int | None
    """Memory limit in force while the execution ran, in bytes."""

    cpu_limit: float | None
    """CPU cores available to the execution."""

    resources_exclusive: bool | None
    """Whether this execution was the only one running on the worker."""

    queue_seconds: float | None
    """Time between submission and the start of the execution, in seconds."""

    created_at: Any
    """Timestamp the execution was created."""

    @property
    def parallelism(self) -> float | None:
        """Mean core occupancy, or ``None`` when either input is missing."""
        if self.cpu_seconds is None or not self.wall_seconds:
            return None
        return self.cpu_seconds / self.wall_seconds

cpu_limit instance-attribute

CPU cores available to the execution.

cpu_seconds instance-attribute

CPU time consumed by the execution and its children, in seconds.

created_at instance-attribute

Timestamp the execution was created.

diagnostic_slug instance-attribute

Owning diagnostic's slug.

execution_id instance-attribute

Primary key of the underlying Execution row.

memory_limit_bytes instance-attribute

Memory limit in force while the execution ran, in bytes.

memory_source instance-attribute

Provenance of peak_memory_bytes.

parallelism property

Mean core occupancy, or None when either input is missing.

peak_memory_bytes instance-attribute

Peak resident memory observed during the execution, in bytes.

provider_slug instance-attribute

Owning provider's slug.

queue_seconds instance-attribute

Time between submission and the start of the execution, in seconds.

resources_exclusive instance-attribute

Whether this execution was the only one running on the worker.

successful instance-attribute

True/False once the execution has finished, None while still running.

wall_seconds instance-attribute

Wall clock time taken by the execution, in seconds.

ResourceProfile

Aggregated resource usage for one diagnostic, or for one provider as a roll-up.

Every percentile is computed over the usable samples only. A usable sample is an execution that recorded a wall time, a CPU time and a peak memory figure, whose memory source matches memory_source, and whose peak is attributable to it alone when exclusive_only was requested, which only rules out a cgroup reading taken from a shared container. Executions that recorded nothing at all are absent from both counts, because an unmeasured run is not a run that used no memory.

A run that failed is aggregated like any other, provided it recorded a measurement. A diagnostic that reached 40 GiB before dying is the row that most needs to reach a recommendation.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

@attrs.frozen(kw_only=True)
class ResourceProfile:
    """
    Aggregated resource usage for one diagnostic, or for one provider as a roll-up.

    Every percentile is computed over the *usable samples* only.
    A usable sample is an execution that recorded a wall time, a CPU time and a peak memory figure,
    whose memory source matches ``memory_source``,
    and whose peak is attributable to it alone when ``exclusive_only`` was requested,
    which only rules out a cgroup reading taken from a shared container.
    Executions that recorded nothing at all are absent from both counts,
    because an unmeasured run is not a run that used no memory.

    A run that failed is aggregated like any other, provided it recorded a measurement.
    A diagnostic that reached 40 GiB before dying is the row that most needs to reach a recommendation.
    """

    provider_slug: str
    """Owning provider's slug."""

    diagnostic_slug: str | None
    """Owning diagnostic's slug, or ``None`` for a provider-level roll-up."""

    n_samples: int
    """Executions with a usable measurement."""

    n_excluded: int
    """
    Executions that were measured but not usable.

    A row lands here when it carried a cgroup reading from a container it did not have to itself,
    when it was incomplete,
    or when it carried a memory source other than ``memory_source``.
    """

    n_failed: int
    """
    Failed executions in the same window, measured or not.

    An execution killed for exceeding its memory limit records nothing,
    so it never appears in ``n_samples``.
    That biases every recommendation here low on exactly the diagnostics that need more memory,
    and this count is the only warning of it.
    """

    memory_source: MemorySource | None
    """
    Provenance of every peak memory figure aggregated here, or ``None`` when there are no samples.

    A cgroup reading and a rusage reading for the same run can differ by a factor of two,
    so they are never mixed.
    When a group's executions carry more than one source,
    the source contributing the most samples wins and the rest land in ``n_excluded``.
    """

    peak_memory_p50: int
    """Median peak resident memory, in bytes."""

    peak_memory_p95: int
    """95th percentile peak resident memory, in bytes."""

    peak_memory_max: int
    """Largest peak resident memory observed, in bytes."""

    wall_p95: float
    """95th percentile wall clock time, in seconds."""

    cpu_seconds_p95: float
    """95th percentile CPU time, in seconds."""

    parallelism_p95: float
    """
    95th percentile of ``cpu_seconds / wall_seconds``.

    This is the core-count signal.
    A diagnostic sitting near 1.0 is serial and gains nothing from more cores.
    """

    memory_limit_seen: int | None
    """
    Memory limit in force for the most recent sample that recorded one.

    The most recent limit rather than the largest, because the question being answered is
    whether the container as it is configured now is big enough.
    """

    headroom_ratio: float | None
    """
    ``memory_limit_seen / peak_memory_p95``, or ``None`` when no limit was recorded.

    Below 1 means the diagnostic is being killed by the current container size.
    """

    safety_factor: float = 1.3
    """Multiplier applied to the p95 peak when recommending a memory size."""

    @property
    def recommended_memory_bytes(self) -> int:
        """Peak memory p95 scaled by ``safety_factor``, rounded up to a whole GiB."""
        scaled = self.peak_memory_p95 * self.safety_factor
        return max(1, math.ceil(scaled / _BYTES_PER_GIB)) * _BYTES_PER_GIB

    @property
    def recommended_cpus(self) -> int:
        """Mean parallelism at the 95th percentile, rounded up, never below one core."""
        return max(1, math.ceil(self.parallelism_p95))

    @property
    def confidence(self) -> Confidence:
        """
        How much weight the recommendation can carry.

        ``none`` when there are no usable samples at all,
        ``thin`` below ten samples, where the p95 is indistinguishable from the maximum,
        and ``good`` at or above ten.
        """
        if self.n_samples == 0:
            return "none"
        if self.n_samples < _GOOD_SAMPLE_COUNT:
            return "thin"
        return "good"

    @property
    def over_limit(self) -> bool:
        """Whether the p95 peak has outgrown the recorded memory limit."""
        return self.headroom_ratio is not None and self.headroom_ratio < 1.0

confidence property

How much weight the recommendation can carry.

none when there are no usable samples at all, thin below ten samples, where the p95 is indistinguishable from the maximum, and good at or above ten.

cpu_seconds_p95 instance-attribute

95th percentile CPU time, in seconds.

diagnostic_slug instance-attribute

Owning diagnostic's slug, or None for a provider-level roll-up.

headroom_ratio instance-attribute

memory_limit_seen / peak_memory_p95, or None when no limit was recorded.

Below 1 means the diagnostic is being killed by the current container size.

memory_limit_seen instance-attribute

Memory limit in force for the most recent sample that recorded one.

The most recent limit rather than the largest, because the question being answered is whether the container as it is configured now is big enough.

memory_source instance-attribute

Provenance of every peak memory figure aggregated here, or None when there are no samples.

A cgroup reading and a rusage reading for the same run can differ by a factor of two, so they are never mixed. When a group's executions carry more than one source, the source contributing the most samples wins and the rest land in n_excluded.

n_excluded instance-attribute

Executions that were measured but not usable.

A row lands here when it carried a cgroup reading from a container it did not have to itself, when it was incomplete, or when it carried a memory source other than memory_source.

n_failed instance-attribute

Failed executions in the same window, measured or not.

An execution killed for exceeding its memory limit records nothing, so it never appears in n_samples. That biases every recommendation here low on exactly the diagnostics that need more memory, and this count is the only warning of it.

n_samples instance-attribute

Executions with a usable measurement.

over_limit property

Whether the p95 peak has outgrown the recorded memory limit.

parallelism_p95 instance-attribute

95th percentile of cpu_seconds / wall_seconds.

This is the core-count signal. A diagnostic sitting near 1.0 is serial and gains nothing from more cores.

peak_memory_max instance-attribute

Largest peak resident memory observed, in bytes.

peak_memory_p50 instance-attribute

Median peak resident memory, in bytes.

peak_memory_p95 instance-attribute

95th percentile peak resident memory, in bytes.

provider_slug instance-attribute

Owning provider's slug.

recommended_cpus property

Mean parallelism at the 95th percentile, rounded up, never below one core.

recommended_memory_bytes property

Peak memory p95 scaled by safety_factor, rounded up to a whole GiB.

safety_factor = 1.3 class-attribute instance-attribute

Multiplier applied to the p95 peak when recommending a memory size.

wall_p95 instance-attribute

95th percentile wall clock time, in seconds.

ResourcesReader

Per-execution resource measurement read domain.

Constructed from a Database, which owns the session and the read-only story. All read methods return detached DTOs that outlive the session.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

class ResourcesReader:
    """
    Per-execution resource measurement read domain.

    Constructed from a [Database][climate_ref.database.Database],
    which owns the session and the read-only story.
    All read methods return detached DTOs that outlive the session.
    """

    def __init__(self, database: Database, *, session: Session | None = None) -> None:
        self._db = database
        self._session = session

    @property
    def session(self) -> Session:
        """The underlying database session."""
        return self._session if self._session is not None else self._db.session

    def _to_view(self, row: Any) -> ResourceMeasurementView:
        return ResourceMeasurementView(
            execution_id=row.execution_id,
            provider_slug=row.provider_slug,
            diagnostic_slug=row.diagnostic_slug,
            successful=row.successful,
            wall_seconds=row.wall_seconds,
            cpu_seconds=row.cpu_seconds,
            peak_memory_bytes=row.peak_memory_bytes,
            memory_source=row.memory_source,
            memory_limit_bytes=row.memory_limit_bytes,
            cpu_limit=row.cpu_limit,
            resources_exclusive=row.resources_exclusive,
            queue_seconds=row.queue_seconds,
            created_at=row.created_at,
        )

    def measurements(
        self,
        filters: ResourceFilter | None = None,
        *,
        offset: int = 0,
        limit: int | None = None,
    ) -> ResourceMeasurementCollection:
        """
        Query the raw per-execution measurements behind the profiles.

        Pagination is applied in SQL over the deterministic ``created_at, id`` ordering.

        Parameters
        ----------
        filters
            Restricts the executions considered. ``None`` means no restriction.
        offset
            Rows to skip before the returned page.
        limit
            Page size, or ``None`` for every matching row.

        Returns
        -------
        :
            A page of measurements plus the pre-pagination total.
        """
        stmt = select_execution_resources(filters)
        total_count = count_values(self.session, stmt)

        page = stmt.offset(offset)
        if limit is not None:
            page = page.limit(limit)

        items = tuple(self._to_view(row) for row in self.session.execute(page).all())
        return ResourceMeasurementCollection(items=items, total_count=total_count, offset=offset, limit=limit)

    def measurement(self, execution_id: int) -> ResourceMeasurementView | None:
        """
        Fetch one execution's measurement by execution id.

        Returns ``None`` when no execution has that id,
        or when the execution neither recorded a measurement nor failed.
        """
        stmt = select_execution_resources().where(Execution.id == execution_id)
        row = self.session.execute(stmt).one_or_none()
        return self._to_view(row) if row is not None else None

    def profiles(  # noqa: PLR0913
        self,
        *,
        diagnostic_contains: Sequence[str] | None = None,
        provider_contains: Sequence[str] | None = None,
        since: datetime.datetime | None = None,
        group_by: GroupBy = "diagnostic",
        exclusive_only: bool = True,
        safety_factor: float = 1.3,
    ) -> tuple[ResourceProfile, ...]:
        """
        Aggregate resource measurements into a sizing answer per diagnostic or per provider.

        Rows are filtered in SQL and the percentiles are computed in Python,
        because SQLite has no percentile function.

        A cgroup peak recorded while a worker ran four executions at once measures the worker,
        not the execution, so ``exclusive_only`` defaults to ``True``.
        Those rows are counted in ``n_excluded`` rather than dropped silently.
        It costs nothing on the other sources,
        which sweep the execution's own processes and stay attributable however busy the worker was.

        Failed executions that recorded a measurement are aggregated alongside successful ones.
        A run that died at 40 GiB is the strongest evidence there is about what the diagnostic needs.

        Peaks from different memory sources are never mixed.
        Within each group the source contributing the most samples wins,
        the rest are counted in ``n_excluded``,
        and the winner is reported as ``memory_source``.

        Parameters
        ----------
        diagnostic_contains
            Case-insensitive substring matches on diagnostic slug (OR-combined).
        provider_contains
            Case-insensitive substring matches on provider slug (OR-combined).
        since
            Keep only executions created at or after this naive UTC timestamp.
        group_by
            ``diagnostic`` for one profile per diagnostic,
            or ``provider`` for the worker-sizing roll-up across a provider's diagnostics.
        exclusive_only
            Drop cgroup readings taken from a container the execution did not have to itself.
        safety_factor
            Multiplier applied to the p95 peak by ``recommended_memory_bytes``.

        Returns
        -------
        :
            One profile per group, ordered by provider then diagnostic slug.
        """
        filters = ResourceFilter(
            diagnostic_contains=diagnostic_contains,
            provider_contains=provider_contains,
            since=since,
        )
        rows = [self._to_view(row) for row in self.session.execute(select_execution_resources(filters)).all()]

        grouped: dict[tuple[str, str], list[ResourceMeasurementView]] = {}
        for row in rows:
            grouped.setdefault((row.provider_slug, row.diagnostic_slug), []).append(row)

        per_diagnostic = [
            _build_profile(
                provider_slug=provider_slug,
                diagnostic_slug=diagnostic_slug,
                rows=group_rows,
                exclusive_only=exclusive_only,
                safety_factor=safety_factor,
            )
            for (provider_slug, diagnostic_slug), group_rows in sorted(grouped.items())
        ]

        if group_by == "diagnostic":
            return tuple(per_diagnostic)

        by_provider: dict[str, list[ResourceProfile]] = {}
        for profile in per_diagnostic:
            by_provider.setdefault(profile.provider_slug, []).append(profile)
        return tuple(_roll_up(slug, profiles) for slug, profiles in sorted(by_provider.items()))

session property

The underlying database session.

measurement(execution_id)

Fetch one execution's measurement by execution id.

Returns None when no execution has that id, or when the execution neither recorded a measurement nor failed.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

def measurement(self, execution_id: int) -> ResourceMeasurementView | None:
    """
    Fetch one execution's measurement by execution id.

    Returns ``None`` when no execution has that id,
    or when the execution neither recorded a measurement nor failed.
    """
    stmt = select_execution_resources().where(Execution.id == execution_id)
    row = self.session.execute(stmt).one_or_none()
    return self._to_view(row) if row is not None else None

measurements(filters=None, *, offset=0, limit=None)

Query the raw per-execution measurements behind the profiles.

Pagination is applied in SQL over the deterministic created_at, id ordering.

Parameters:

Name	Type	Description	Default
filters	ResourceFilter | None	Restricts the executions considered. None means no restriction.	None
offset	int	Rows to skip before the returned page.	0
limit	int | None	Page size, or None for every matching row.	None

Returns:

Type	Description
ResourceMeasurementCollection	A page of measurements plus the pre-pagination total.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

def measurements(
    self,
    filters: ResourceFilter | None = None,
    *,
    offset: int = 0,
    limit: int | None = None,
) -> ResourceMeasurementCollection:
    """
    Query the raw per-execution measurements behind the profiles.

    Pagination is applied in SQL over the deterministic ``created_at, id`` ordering.

    Parameters
    ----------
    filters
        Restricts the executions considered. ``None`` means no restriction.
    offset
        Rows to skip before the returned page.
    limit
        Page size, or ``None`` for every matching row.

    Returns
    -------
    :
        A page of measurements plus the pre-pagination total.
    """
    stmt = select_execution_resources(filters)
    total_count = count_values(self.session, stmt)

    page = stmt.offset(offset)
    if limit is not None:
        page = page.limit(limit)

    items = tuple(self._to_view(row) for row in self.session.execute(page).all())
    return ResourceMeasurementCollection(items=items, total_count=total_count, offset=offset, limit=limit)

profiles(*, diagnostic_contains=None, provider_contains=None, since=None, group_by='diagnostic', exclusive_only=True, safety_factor=1.3)

Aggregate resource measurements into a sizing answer per diagnostic or per provider.

Rows are filtered in SQL and the percentiles are computed in Python, because SQLite has no percentile function.

A cgroup peak recorded while a worker ran four executions at once measures the worker, not the execution, so exclusive_only defaults to True. Those rows are counted in n_excluded rather than dropped silently. It costs nothing on the other sources, which sweep the execution's own processes and stay attributable however busy the worker was.

Failed executions that recorded a measurement are aggregated alongside successful ones. A run that died at 40 GiB is the strongest evidence there is about what the diagnostic needs.

Peaks from different memory sources are never mixed. Within each group the source contributing the most samples wins, the rest are counted in n_excluded, and the winner is reported as memory_source.

Parameters:

Name	Type	Description	Default
diagnostic_contains	Sequence[str] | None	Case-insensitive substring matches on diagnostic slug (OR-combined).	None
provider_contains	Sequence[str] | None	Case-insensitive substring matches on provider slug (OR-combined).	None
since	datetime | None	Keep only executions created at or after this naive UTC timestamp.	None
group_by	GroupBy	diagnostic for one profile per diagnostic, or provider for the worker-sizing roll-up across a provider's diagnostics.	'diagnostic'
exclusive_only	bool	Drop cgroup readings taken from a container the execution did not have to itself.	True
safety_factor	float	Multiplier applied to the p95 peak by recommended_memory_bytes.	1.3

Returns:

Type	Description
tuple[ResourceProfile, ...]	One profile per group, ordered by provider then diagnostic slug.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

def profiles(  # noqa: PLR0913
    self,
    *,
    diagnostic_contains: Sequence[str] | None = None,
    provider_contains: Sequence[str] | None = None,
    since: datetime.datetime | None = None,
    group_by: GroupBy = "diagnostic",
    exclusive_only: bool = True,
    safety_factor: float = 1.3,
) -> tuple[ResourceProfile, ...]:
    """
    Aggregate resource measurements into a sizing answer per diagnostic or per provider.

    Rows are filtered in SQL and the percentiles are computed in Python,
    because SQLite has no percentile function.

    A cgroup peak recorded while a worker ran four executions at once measures the worker,
    not the execution, so ``exclusive_only`` defaults to ``True``.
    Those rows are counted in ``n_excluded`` rather than dropped silently.
    It costs nothing on the other sources,
    which sweep the execution's own processes and stay attributable however busy the worker was.

    Failed executions that recorded a measurement are aggregated alongside successful ones.
    A run that died at 40 GiB is the strongest evidence there is about what the diagnostic needs.

    Peaks from different memory sources are never mixed.
    Within each group the source contributing the most samples wins,
    the rest are counted in ``n_excluded``,
    and the winner is reported as ``memory_source``.

    Parameters
    ----------
    diagnostic_contains
        Case-insensitive substring matches on diagnostic slug (OR-combined).
    provider_contains
        Case-insensitive substring matches on provider slug (OR-combined).
    since
        Keep only executions created at or after this naive UTC timestamp.
    group_by
        ``diagnostic`` for one profile per diagnostic,
        or ``provider`` for the worker-sizing roll-up across a provider's diagnostics.
    exclusive_only
        Drop cgroup readings taken from a container the execution did not have to itself.
    safety_factor
        Multiplier applied to the p95 peak by ``recommended_memory_bytes``.

    Returns
    -------
    :
        One profile per group, ordered by provider then diagnostic slug.
    """
    filters = ResourceFilter(
        diagnostic_contains=diagnostic_contains,
        provider_contains=provider_contains,
        since=since,
    )
    rows = [self._to_view(row) for row in self.session.execute(select_execution_resources(filters)).all()]

    grouped: dict[tuple[str, str], list[ResourceMeasurementView]] = {}
    for row in rows:
        grouped.setdefault((row.provider_slug, row.diagnostic_slug), []).append(row)

    per_diagnostic = [
        _build_profile(
            provider_slug=provider_slug,
            diagnostic_slug=diagnostic_slug,
            rows=group_rows,
            exclusive_only=exclusive_only,
            safety_factor=safety_factor,
        )
        for (provider_slug, diagnostic_slug), group_rows in sorted(grouped.items())
    ]

    if group_by == "diagnostic":
        return tuple(per_diagnostic)

    by_provider: dict[str, list[ResourceProfile]] = {}
    for profile in per_diagnostic:
        by_provider.setdefault(profile.provider_slug, []).append(profile)
    return tuple(_roll_up(slug, profiles) for slug, profiles in sorted(by_provider.items()))

select_execution_resources(filters=None)

Build the Select over the resource columns on Execution.

Rows are kept when they carry a measurement or when they failed. A failed execution with no measurement is what an out-of-memory kill looks like, and that count is needed to report the bias it introduces.

Ordered by created_at, id ascending so SQL pagination is deterministic across pages.

Parameters:

Name	Type	Description	Default
filters	ResourceFilter | None	Restricts the executions considered. None means no restriction.	None

Returns:

Type	Description
Select[Any]	A Select yielding one row per execution.

Source code in packages/climate-ref/src/climate_ref/results/resources.py

def select_execution_resources(filters: ResourceFilter | None = None) -> Select[Any]:
    """
    Build the ``Select`` over the resource columns on ``Execution``.

    Rows are kept when they carry a measurement or when they failed.
    A failed execution with no measurement is what an out-of-memory kill looks like,
    and that count is needed to report the bias it introduces.

    Ordered by ``created_at, id`` ascending so SQL pagination is deterministic across pages.

    Parameters
    ----------
    filters
        Restricts the executions considered. ``None`` means no restriction.

    Returns
    -------
    :
        A ``Select`` yielding one row per execution.
    """
    filters = filters or ResourceFilter()

    stmt = (
        select(
            Execution.id.label("execution_id"),
            Provider.slug.label("provider_slug"),
            Diagnostic.slug.label("diagnostic_slug"),
            Execution.successful,
            Execution.wall_seconds,
            Execution.cpu_seconds,
            Execution.peak_memory_bytes,
            Execution.memory_source,
            Execution.memory_limit_bytes,
            Execution.cpu_limit,
            Execution.resources_exclusive,
            Execution.queue_seconds,
            Execution.created_at,
        )
        .join(ExecutionGroup, Execution.execution_group_id == ExecutionGroup.id)
        .join(Diagnostic, ExecutionGroup.diagnostic_id == Diagnostic.id)
        .join(Provider, Diagnostic.provider_id == Provider.id)
        .where(
            or_(
                Execution.wall_seconds.is_not(None),
                Execution.peak_memory_bytes.is_not(None),
                Execution.successful.is_(False),
            )
        )
        .order_by(Execution.created_at, Execution.id)
    )

    if filters.diagnostic_contains:
        stmt = stmt.where(
            or_(*(Diagnostic.slug.ilike(f"%{s.lower()}%") for s in filters.diagnostic_contains))
        )
    if filters.provider_contains:
        stmt = stmt.where(or_(*(Provider.slug.ilike(f"%{s.lower()}%") for s in filters.provider_contains)))
    if filters.since is not None:
        stmt = stmt.where(Execution.created_at >= filters.since)

    return stmt