climate_ref.executor.result_handling

Execute diagnostics in different environments

We support running diagnostics in different environments, such as locally, in a separate process, or in a container. These environments are represented by climate_ref.executor.Executor classes.

The simplest executor is the LocalExecutor, which runs the diagnostic in the same process. This is useful for local testing and debugging.

ExecutionFuture

A container linking a submitted future to its execution metadata.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

@define
class ExecutionFuture:
    """A container linking a submitted future to its execution metadata."""

    future: Future[ExecutionResult]
    """
    The future representing the asynchronous execution of this task.

    This future has a base-class of ``concurrent.futures.Future``,
    but the concrete class of the instance will depend on the executor.
    """

    definition: ExecutionDefinition
    """
    The execution definition associated with this future.
    """

    execution_id: int | None = None
    """
    The ID of the execution associated with this future, or ``None`` if not yet assigned.
    """

    submitted_at: float = 0.0
    """
    Wall-clock time (``time.time()``) at which this future was submitted to the executor.
    """

    started_at: float | None = None
    """
    Wall-clock time (``time.time()``) at which a worker was first observed running this future,
    or ``None`` while it is still queued.

    The per-task timeout is budgeted against this rather than ``submitted_at``
    so that time spent waiting in the pool queue is not counted as execution time.
    """

    def __repr__(self) -> str:
        return f"<ExecutionFuture {self.definition.execution_slug()} execution_id={self.execution_id}>"

    @property
    def queue_seconds(self) -> float | None:
        """
        Time spent waiting between submission and the start of execution, in seconds.

        ``None`` when the start of execution could not be established.
        """
        if self.started_at is None:
            return None
        return max(0.0, self.started_at - self.submitted_at)

    def infer_started_at(self, result: ExecutionResult, observed_at: float) -> None:
        """
        Place ``started_at`` using the worker's own wall time, if a poll never observed the start.

        A task submitted and finished between two polls is never seen running,
        which would leave the queue latency unrecorded for every short execution.
        The worker measured how long it ran,
        so subtracting that from the moment completion was observed puts the start
        within one poll interval of the truth.

        Parameters
        ----------
        result
            The finished result, whose ``resource_usage`` carries the measured wall time.
        observed_at
            Wall-clock time (``time.time()``) at which the future was seen to be done.
        """
        if self.started_at is not None:
            return
        usage = result.resource_usage
        if usage is None:
            return
        self.started_at = max(self.submitted_at, observed_at - usage.wall_seconds)

definition instance-attribute

The execution definition associated with this future.

execution_id = None class-attribute instance-attribute

The ID of the execution associated with this future, or None if not yet assigned.

future instance-attribute

The future representing the asynchronous execution of this task.

This future has a base-class of concurrent.futures.Future, but the concrete class of the instance will depend on the executor.

queue_seconds property

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

None when the start of execution could not be established.

started_at = None class-attribute instance-attribute

Wall-clock time (time.time()) at which a worker was first observed running this future, or None while it is still queued.

The per-task timeout is budgeted against this rather than submitted_at so that time spent waiting in the pool queue is not counted as execution time.

submitted_at = 0.0 class-attribute instance-attribute

Wall-clock time (time.time()) at which this future was submitted to the executor.

infer_started_at(result, observed_at)

Place started_at using the worker's own wall time, if a poll never observed the start.

A task submitted and finished between two polls is never seen running, which would leave the queue latency unrecorded for every short execution. The worker measured how long it ran, so subtracting that from the moment completion was observed puts the start within one poll interval of the truth.

Parameters:

Name	Type	Description	Default
result	ExecutionResult	The finished result, whose resource_usage carries the measured wall time.	required
observed_at	float	Wall-clock time (time.time()) at which the future was seen to be done.	required

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def infer_started_at(self, result: ExecutionResult, observed_at: float) -> None:
    """
    Place ``started_at`` using the worker's own wall time, if a poll never observed the start.

    A task submitted and finished between two polls is never seen running,
    which would leave the queue latency unrecorded for every short execution.
    The worker measured how long it ran,
    so subtracting that from the moment completion was observed puts the start
    within one poll interval of the truth.

    Parameters
    ----------
    result
        The finished result, whose ``resource_usage`` carries the measured wall time.
    observed_at
        Wall-clock time (``time.time()``) at which the future was seen to be done.
    """
    if self.started_at is not None:
        return
    usage = result.resource_usage
    if usage is None:
        return
    self.started_at = max(self.submitted_at, observed_at - usage.wall_seconds)

fail_future(database, config, future, *, detail)

Mark the execution behind an abandoned future as failed-retryable.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def fail_future(database: Database, config: "Config", future: ExecutionFuture, *, detail: str) -> None:
    """Mark the execution behind an abandoned future as failed-retryable."""
    mark_execution_failed(
        database, config, future.definition, future.execution_id, retryable=True, detail=detail
    )

fail_outstanding(database, config, results, progress, *, detail)

Mark every execution an executor is giving up on as failed-retryable and drop it from results.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def fail_outstanding(
    database: Database, config: "Config", results: list[ExecutionFuture], progress: Any, *, detail: str
) -> None:
    """Mark every execution an executor is giving up on as failed-retryable and drop it from ``results``."""
    for outstanding in list(results):
        logger.warning(
            f"Execution {outstanding.definition.execution_slug()!r} was not collected. "
            "Marking it failed-retryable"
        )
        fail_future(database, config, outstanding, detail=detail)
        progress.update(n=1)
        results.remove(outstanding)

handle_execution_result(config, database, execution, result, *, update_dirty=True, queue_seconds=None, detail=None)

Handle the result of a diagnostic execution

This will update the diagnostic execution result with the output of the diagnostic execution. The output will be copied from the scratch directory to the executions directory.

Parameters:

Name	Type	Description	Default
config	Config	The configuration to use	required
database	Database	The active database session to use	required
execution	Execution	The diagnostic execution result DB object to update	required
result	ExecutionResult	The result of the diagnostic execution, either successful or failed	required
update_dirty	bool	Whether to update the execution group's dirty flag. Set to False for reingest which should not alter pending-work state.	True
queue_seconds	float | None	Submit-to-start latency observed by the submitting executor, or None when it is unknown.	None
detail	str | None	Why the executor failed the execution, if it did. A failed execution has this appended to its results log.	None

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def handle_execution_result(  # noqa: PLR0913
    config: "Config",
    database: Database,
    execution: Execution,
    result: "ExecutionResult",
    *,
    update_dirty: bool = True,
    queue_seconds: float | None = None,
    detail: str | None = None,
) -> None:
    """
    Handle the result of a diagnostic execution

    This will update the diagnostic execution result with the output of the diagnostic execution.
    The output will be copied from the scratch directory to the executions directory.

    Parameters
    ----------
    config
        The configuration to use
    database
        The active database session to use
    execution
        The diagnostic execution result DB object to update
    result
        The result of the diagnostic execution, either successful or failed
    update_dirty
        Whether to update the execution group's dirty flag.
        Set to False for reingest which should not alter pending-work state.
    queue_seconds
        Submit-to-start latency observed by the submitting executor, or None when it is unknown.
    detail
        Why the executor failed the execution, if it did.
        A failed execution has this appended to its results log.
    """
    # Recorded before anything that can return early.
    # A run that ran out of memory or blew up during ingest is exactly the run whose numbers are wanted.
    record_resource_usage(execution, result.resource_usage, queue_seconds)

    # Always copy log data to the results directory
    try:
        copy_output_file(
            config.paths.scratch,
            config.paths.results,
            execution.output_fragment,
            EXECUTION_LOG_FILENAME,
        )
    except FileNotFoundError:
        # An execution abandoned before a worker picked it up has no output directory,
        # so a missing log is expected rather than a sign that something went wrong.
        started = (config.paths.scratch / execution.output_fragment).exists()
        report = logger.error if started else logger.debug
        report(
            f"Could not find log file {EXECUTION_LOG_FILENAME} in scratch directory: {config.paths.scratch}. "
            f"This is likely a system error (will be retried on next solve)."
        )
        record_failure(
            config,
            execution,
            ExecutionFailureReason.SystemError if started else ExecutionFailureReason.NotStarted,
            detail or "no log file was written, the process was likely killed",
        )
        if update_dirty:
            # Missing log file suggests the process was killed before writing output,
            # so set dirty=True for retry rather than assuming it was already set.
            execution.execution_group.dirty = True
        return

    if not result.successful or result.metric_bundle_filename is None:
        record_failure(
            config,
            execution,
            ExecutionFailureReason.SystemError
            if result.retryable
            else ExecutionFailureReason.DiagnosticError,
            detail or "the diagnostic reported failure",
        )
        if result.retryable:
            logger.error(f"{execution} failed due to a system error (will be retried on next solve)")
            if update_dirty:
                # A hash-change rerun starts from dirty=False, so set it explicitly
                # rather than assuming it was already True.
                execution.execution_group.dirty = True
        else:
            logger.error(f"{execution} failed due to a diagnostic error")
            if update_dirty:
                execution.execution_group.dirty = False
        return

    logger.info(f"{execution} successful")

    # Copy the curated subset of outputs that REF persists for a successful execution
    # (metric bundle, output bundle and the files it references, series)
    # from scratch to results.
    # Only the files in results are served by the API.
    try:
        copy_execution_outputs(
            config.paths.scratch,
            config.paths.results,
            execution.output_fragment,
            result,
        )
    except (FileNotFoundError, OSError) as exc:
        # The diagnostic reported success but a file its bundle references is
        # missing or unreadable. Fail this execution rather than aborting the
        # whole result-collection loop.
        logger.exception(f"Could not copy outputs for {execution}")
        record_failure(
            config, execution, ExecutionFailureReason.ResultError, f"could not copy outputs: {exc!r}"
        )
        if update_dirty:
            execution.execution_group.dirty = True
        return

    # Ingest outputs and metrics into the database via the shared ingestion path
    cv = CV.load(config.paths.dimensions_cv_resource)
    try:
        with database.session.begin_nested():
            ingest_execution_result(
                database,
                execution,
                result,
                cv,
                output_base_path=config.paths.scratch / execution.output_fragment,
            )
    except Exception as exc:
        # The diagnostic ran, but persisting its results failed
        # (malformed bundle, CV/schema mismatch, DB error).
        # The savepoint already rolled back the partial inserts.
        # Never report success with no metric values.
        logger.exception("Something went wrong when ingesting execution result")
        record_failure(
            config, execution, ExecutionFailureReason.ResultError, f"could not ingest results: {exc!r}"
        )
        if update_dirty:
            execution.execution_group.dirty = True
        return

    # TODO: This should check if the result is the most recent for the execution,
    # if so then update the dirty fields
    # i.e. if there are outstanding executions don't make as clean
    if update_dirty:
        execution.execution_group.dirty = False

    # Finally, mark the execution as successful
    execution.mark_successful(result.as_relative_path(result.metric_bundle_filename))

ingest_execution_result(database, execution, result, cv, *, output_base_path)

Ingest a successful execution result into the database.

Registers output entries and ingests scalar and series metric values.

Parameters:

Name	Type	Description	Default
database	Database	The active database session to use	required
execution	Execution	The execution record to associate results with	required
result	ExecutionResult	The successful execution result	required
cv	CV	The controlled vocabulary to validate metrics against	required
output_base_path	Path	Primary base directory for resolving output filenames	required

Notes

Callers are responsible for:

• File copying (scratch -> results)
• Transaction boundaries
• Marking the execution as successful (execution.mark_successful())
• Setting the dirty flag on the execution group

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def ingest_execution_result(
    database: Database,
    execution: Execution,
    result: "ExecutionResult",
    cv: CV,
    *,
    output_base_path: pathlib.Path,
) -> None:
    """
    Ingest a successful execution result into the database.

    Registers output entries and ingests scalar and series metric values.

    Parameters
    ----------
    database
        The active database session to use
    execution
        The execution record to associate results with
    result
        The successful execution result
    cv
        The controlled vocabulary to validate metrics against
    output_base_path
        Primary base directory for resolving output filenames

    Notes
    -----
    Callers are responsible for:

    * File copying (scratch -> results)
    * Transaction boundaries
    * Marking the execution as successful (``execution.mark_successful()``)
    * Setting the dirty flag on the execution group
    """
    if result.output_bundle_filename:
        cmec_output_bundle = CMECOutput.load_from_json(result.to_output_path(result.output_bundle_filename))
        for attr, output_type in [
            ("plots", ResultOutputType.Plot),
            ("data", ResultOutputType.Data),
            ("html", ResultOutputType.HTML),
        ]:
            register_execution_outputs(
                database,
                execution,
                getattr(cmec_output_bundle, attr),
                output_type=output_type,
                base_path=output_base_path,
            )

    if result.series_filename:
        ingest_series_values(
            database=database,
            result=result,
            execution=execution,
            cv=cv,
        )

    ingest_scalar_values(
        database=database,
        result=result,
        execution=execution,
        cv=cv,
    )

ingest_scalar_values(database, result, execution, cv)

Load, validate, and bulk-insert scalar metric values.

Parameters:

Name	Type	Description	Default
database	Database	The active database session to use	required
result	ExecutionResult	The execution result containing the metric bundle filename	required
execution	Execution	The execution record to associate values with	required
cv	CV	The controlled vocabulary to validate against	required

Notes

Callers are responsible for transaction boundaries; this function does not open a nested transaction or catch exceptions.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def ingest_scalar_values(
    database: Database,
    result: "ExecutionResult",
    execution: Execution,
    cv: CV,
) -> None:
    """
    Load, validate, and bulk-insert scalar metric values.

    Parameters
    ----------
    database
        The active database session to use
    result
        The execution result containing the metric bundle filename
    execution
        The execution record to associate values with
    cv
        The controlled vocabulary to validate against

    Notes
    -----
    Callers are responsible for transaction boundaries; this function does not
    open a nested transaction or catch exceptions.
    """
    cmec_metric_bundle = CMECMetric.load_from_json(result.to_output_path(result.metric_bundle_filename))

    try:
        cv.validate_metrics(cmec_metric_bundle)
    except (ResultValidationError, AssertionError):
        # TODO: Remove once we have settled on a controlled vocabulary
        logger.opt(exception=True).warning(
            "Diagnostic scalar values do not conform with the controlled vocabulary"
        )

    new_values = []
    for metric_result in cmec_metric_bundle.iter_results():
        kind, dimensions = _validated_kind_and_dimensions(metric_result)
        new_values.append(
            {
                "execution_id": execution.id,
                "value": metric_result.value,
                "attributes": metric_result.attributes,
                "kind": kind,
                **dimensions,
            }
        )

    logger.debug(f"Ingesting {len(new_values)} scalar values for execution {execution.id}")

    if new_values:
        database.session.execute(insert(ScalarMetricValue), new_values)

ingest_series_values(database, result, execution, cv)

Load, validate, and bulk-insert series metric values.

Parameters:

Name	Type	Description	Default
database	Database	The active database session to use	required
result	ExecutionResult	The execution result containing the series filename	required
execution	Execution	The execution record to associate values with	required
cv	CV	The controlled vocabulary to validate against	required

Notes

Callers are responsible for transaction boundaries; this function does not open a nested transaction or catch exceptions.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def ingest_series_values(
    database: Database,
    result: "ExecutionResult",
    execution: Execution,
    cv: CV,
) -> None:
    """
    Load, validate, and bulk-insert series metric values.

    Parameters
    ----------
    database
        The active database session to use
    result
        The execution result containing the series filename
    execution
        The execution record to associate values with
    cv
        The controlled vocabulary to validate against

    Notes
    -----
    Callers are responsible for transaction boundaries; this function does not
    open a nested transaction or catch exceptions.
    """
    assert result.series_filename, "Series filename must be set in the result"

    series_values_path = result.to_output_path(result.series_filename)
    series_values = TSeries.load_from_json(series_values_path)

    try:
        cv.validate_metrics(series_values)
    except (ResultValidationError, AssertionError):
        # TODO: Remove once we have settled on a controlled vocabulary
        logger.opt(exception=True).warning(
            "Diagnostic series values do not conform with the controlled vocabulary"
        )

    # Resolve (deduplicate) the shared index axes for this batch up front,
    # so each distinct index is stored once in ``index_axis``
    # and referenced by id rather than duplicated on every series row.
    digest_by_series = [
        SeriesIndex.compute_hash(series_result.index_name, series_result.index)
        for series_result in series_values
    ]
    axis_payload_by_hash: dict[str, tuple[str | None, Sequence[float | int | str]]] = {}
    for series_result, digest in zip(series_values, digest_by_series, strict=True):
        axis_payload_by_hash.setdefault(digest, (series_result.index_name, series_result.index))

    axis_id_by_hash = SeriesIndex.bulk_get_or_create(database.session, axis_payload_by_hash)

    new_values = []
    for series_result, digest in zip(series_values, digest_by_series, strict=True):
        kind, dimensions = _validated_kind_and_dimensions(series_result)
        row = {
            "execution_id": execution.id,
            "values": series_result.values,
            "attributes": series_result.attributes,
            "index_id": axis_id_by_hash[digest],
            "kind": kind,
            **dimensions,
        }
        if kind == "reference":
            # Stable content hash so an identical observation ingested by different
            # executions deduplicates to the same reference_id.
            row["reference_id"] = SeriesMetricValue.compute_reference_id(
                series_result.values,
                series_result.index,
                dimensions.get("reference_source_id"),
            )
        new_values.append(row)

    logger.debug(f"Ingesting {len(new_values)} series values for execution {execution.id}")

    if new_values:
        database.session.execute(insert(SeriesMetricValue), new_values)

mark_execution_failed(database, config, definition, execution_id, *, retryable, detail=None)

Persist a failed result for an outstanding execution.

Used when an executor abandons a future (per-task timeout, overall timeout, worker crash) so the corresponding Execution row never stays stuck in successful=None state. detail is written to the execution's results log.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def mark_execution_failed(  # noqa: PLR0913
    database: Database,
    config: "Config",
    definition: ExecutionDefinition,
    execution_id: int | None,
    *,
    retryable: bool,
    detail: str | None = None,
) -> None:
    """Persist a failed result for an outstanding execution.

    Used when an executor abandons a future (per-task timeout, overall timeout,
    worker crash) so the corresponding ``Execution`` row never stays stuck in
    ``successful=None`` state.
    ``detail`` is written to the execution's results log.
    """
    try:
        failure_result = ExecutionResult.build_from_failure(definition, retryable=retryable)
        with database.session.begin():
            execution = database.session.get(Execution, execution_id) if execution_id else None
            process_result(config, database, failure_result, execution, detail=detail)
    except Exception:
        logger.exception(f"Failed to record failure for {definition.execution_slug()!r}")
        _force_mark_failed(database, execution_id, retryable=retryable)

process_result(config, database, result, execution, *, queue_seconds=None, detail=None)

Process the result of a diagnostic execution, persisting outcome to the DB.

queue_seconds is supplied by the submitting executor, because only the parent knows when the task was submitted. detail says why a failed execution failed and is written to its results log.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def process_result(  # noqa: PLR0913
    config: "Config",
    database: Database,
    result: ExecutionResult,
    execution: Execution | None,
    *,
    queue_seconds: float | None = None,
    detail: str | None = None,
) -> None:
    """Process the result of a diagnostic execution, persisting outcome to the DB.

    ``queue_seconds`` is supplied by the submitting executor,
    because only the parent knows when the task was submitted.
    ``detail`` says why a failed execution failed and is written to its results log.
    """
    if not result.successful:
        if execution is not None:  # pragma: no branch
            info_msg = (
                f"\nAdditional information about this execution can be viewed using: "
                f"ref executions inspect {execution.execution_group_id}"
            )
        else:
            info_msg = ""
        logger.error(f"Error running {result.definition.execution_slug()}. {info_msg}")

    if execution:
        handle_execution_result(
            config, database, execution, result, queue_seconds=queue_seconds, detail=detail
        )

record_failure(config, execution, reason, detail)

Mark the execution failed and append why to its results log, which is otherwise silent.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def record_failure(
    config: "Config", execution: Execution, reason: ExecutionFailureReason, detail: str
) -> None:
    """Mark the execution failed and append why to its results log, which is otherwise silent."""
    execution.mark_failed(reason)
    log_path = config.paths.results / execution.output_fragment / EXECUTION_LOG_FILENAME
    try:
        log_path.parent.mkdir(parents=True, exist_ok=True)
        with log_path.open("a") as fh:
            fh.write(f"\nREF: execution failed ({reason.value}): {detail}\n")
    except OSError:
        logger.warning(f"Could not write the failure reason to {log_path}", exc_info=True)

record_lost_execution(config, execution, detail)

Fail an execution whose worker died before it returned a result

The partial log is copied from scratch first, so the results log shows how far it got. The group's dirty flag is left alone.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def record_lost_execution(config: "Config", execution: Execution, detail: str) -> None:
    """
    Fail an execution whose worker died before it returned a result

    The partial log is copied from scratch first, so the results log shows how far it got.
    The group's dirty flag is left alone.
    """
    try:
        copy_output_file(
            config.paths.scratch, config.paths.results, execution.output_fragment, EXECUTION_LOG_FILENAME
        )
    except FileNotFoundError:
        logger.debug(f"No partial log for lost execution {execution.id}")
    except (OSError, ValueError) as e:
        logger.warning(f"Could not copy the partial log of lost execution {execution.id}: {e}")
    record_failure(config, execution, ExecutionFailureReason.SystemError, detail)

record_resource_usage(execution, usage, queue_seconds)

Annotate an execution with its resource measurements, swallowing any failure.

These numbers are diagnostic metadata rather than scientific output, so losing a measurement is always preferable to losing a result.

Parameters:

Name	Type	Description	Default
execution	Execution	The execution record to annotate	required
usage	ResourceUsage | None	The measurements taken by the worker, or None when nothing was measured	required
queue_seconds	float | None	Submit-to-start latency observed by the submitting executor, or None when it is unknown	required

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def record_resource_usage(
    execution: Execution,
    usage: ResourceUsage | None,
    queue_seconds: float | None,
) -> None:
    """
    Annotate an execution with its resource measurements, swallowing any failure.

    These numbers are diagnostic metadata rather than scientific output,
    so losing a measurement is always preferable to losing a result.

    Parameters
    ----------
    execution
        The execution record to annotate
    usage
        The measurements taken by the worker, or None when nothing was measured
    queue_seconds
        Submit-to-start latency observed by the submitting executor, or None when it is unknown
    """
    try:
        execution.apply_resource_usage(usage, queue_seconds=queue_seconds)
    except Exception:
        logger.opt(exception=True).warning("Could not record resource usage for this execution")

register_execution_outputs(database, execution, outputs, output_type, *, base_path)

Register output entries in the database.

Each entry in outputs is resolved relative to base_path.

Parameters:

Name	Type	Description	Default
database	Database	The active database session to use	required
execution	Execution	The execution record to associate outputs with	required
outputs	dict[str, OutputDict] | None	Mapping of short name to OutputDict (may be None)	required
output_type	ResultOutputType	The type of output being registered	required
base_path	Path	Base directory for resolving relative filenames	required

Notes

Callers are responsible for transaction boundaries.

Source code in packages/climate-ref/src/climate_ref/executor/result_handling.py

def register_execution_outputs(
    database: Database,
    execution: Execution,
    outputs: "dict[str, OutputDict] | None",
    output_type: ResultOutputType,
    *,
    base_path: pathlib.Path,
) -> None:
    """
    Register output entries in the database.

    Each entry in ``outputs`` is resolved relative to ``base_path``.

    Parameters
    ----------
    database
        The active database session to use
    execution
        The execution record to associate outputs with
    outputs
        Mapping of short name to ``OutputDict`` (may be None)
    output_type
        The type of output being registered
    base_path
        Base directory for resolving relative filenames

    Notes
    -----
    Callers are responsible for transaction boundaries.
    """
    for key, output_info in (outputs or {}).items():
        filename = ensure_relative_path(output_info.filename, base_path)
        database.session.add(
            ExecutionOutput.build(
                execution_id=execution.id,
                output_type=output_type,
                filename=str(filename),
                description=output_info.description,
                short_name=key,
                long_name=output_info.long_name,
                dimensions=output_info.dimensions or {},
            )
        )