quantum_graph_builder.py

# This file is part of pipe_base.
#
# Developed for the LSST Data Management System.
# This product includes software developed by the LSST Project
# (http://www.lsst.org).
# See the COPYRIGHT file at the top-level directory of this distribution
# for details of code ownership.
#
# This software is dual licensed under the GNU General Public License and also
# under a 3-clause BSD license. Recipients may choose which of these licenses
# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
# respectively.  If you choose the GPL option then the following text applies
# (but note that there is still no warranty even if you opt for BSD instead):
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""The base class for the QuantumGraph-generation algorithm and various
helper classes.
"""

from __future__ import annotations

__all__ = (
    "EmptyDimensionsDatasets",
    "OutputExistsError",
    "PrerequisiteMissingError",
    "QuantumGraphBuilder",
    "QuantumGraphBuilderError",
)

import collections
import dataclasses
import operator
from abc import ABC, abstractmethod
from collections import defaultdict
from collections.abc import Iterable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, cast, final

from lsst.daf.butler import (
    Butler,
    CollectionType,
    DataCoordinate,
    DatasetRef,
    DatasetType,
    DimensionDataAttacher,
    DimensionUniverse,
    NamedKeyDict,
    NamedKeyMapping,
    Quantum,
)
from lsst.daf.butler._rubin import generate_uuidv7
from lsst.daf.butler.datastore.record_data import DatastoreRecordData
from lsst.daf.butler.registry import MissingCollectionError, MissingDatasetTypeError
from lsst.daf.butler.utils import globToRegex
from lsst.utils.logging import LsstLogAdapter, getLogger
from lsst.utils.timer import timeMethod

from . import automatic_connection_constants as acc
from ._status import NoWorkFound
from ._task_metadata import TaskMetadata
from .connections import AdjustQuantumHelper, QuantaAdjuster
from .pipeline_graph import Edge, PipelineGraph, TaskNode
from .prerequisite_helpers import PrerequisiteInfo, SkyPixBoundsBuilder, TimespanBuilder
from .quantum_graph_skeleton import (
    DatasetKey,
    PrerequisiteDatasetKey,
    QuantumGraphSkeleton,
    QuantumKey,
    TaskInitKey,
)

if TYPE_CHECKING:
    from .graph import QuantumGraph
    from .pipeline import TaskDef
    from .quantum_graph import PredictedDatasetModel, PredictedQuantumGraphComponents


class QuantumGraphBuilderError(Exception):
    """Base class for exceptions generated by QuantumGraphBuilder."""

    pass


class OutputExistsError(QuantumGraphBuilderError):
    """Exception generated when output datasets already exist."""

    pass


class PrerequisiteMissingError(QuantumGraphBuilderError):
    """Exception generated when a prerequisite dataset does not exist."""

    pass


class InitInputMissingError(QuantumGraphBuilderError):
    """Exception generated when an init-input dataset does not exist."""

    pass


class QuantumGraphBuilder(ABC):
    """An abstract base class for building `.QuantumGraph` objects from a
    pipeline.

    Parameters
    ----------
    pipeline_graph : `.pipeline_graph.PipelineGraph`
        Pipeline to build a `.QuantumGraph` from, as a graph.  Will be resolved
        in-place with the given butler (any existing resolution is ignored).
    butler : `lsst.daf.butler.Butler`
        Client for the data repository.  Should be read-only.
    input_collections : `~collections.abc.Sequence` [ `str` ], optional
        Collections to search for overall-input datasets.  If not provided,
        ``butler.collections`` is used (and must not be empty).
    output_run : `str`, optional
        Output `~lsst.daf.butler.CollectionType.RUN` collection.  If not
        provided, ``butler.run`` is used (and must not be `None`).
    skip_existing_in : `~collections.abc.Sequence` [ `str` ], optional
        Collections to search for outputs that already exist for the purpose of
        skipping quanta that have already been run.
    retained_dataset_types : `~collections.abc.Sequence` [ `str` ], optional
        Dataset type names or glob-style wildcard patterns for dataset types
        that should exist in ``skip_existing_in`` when the producing task ran
        successfully.  When a quantum should run, the builder propagates the
        must-run signal backward through non-retained input datasets, forcing
        the upstream quanta that need to regenerate those intermediates to also
        run.  Has no effect without ``skip_existing_in``.  ``["*"]`` means
        retaining all datasets, equivalent to not providing this option.
    prune_unanchored_quanta : `tuple` [ `str`, `str` ], optional
        A ``(source_label, anchor_label)`` pair of task labels triggering
        unanchored-quantum pruning after the skeleton is assembled.  A
        ``source_label`` quantum is removed along with its entire downstream
        chain if no ``anchor_label`` quantum is reachable from it along
        directed graph edges.
    clobber : `bool`, optional
        Whether to raise if predicted outputs already exist in ``output_run``
        (not including those quanta that would be skipped because they've
        already been run).  This never actually clobbers outputs; it just
        informs the graph generation algorithm whether execution will run with
        clobbering enabled.  This is ignored if ``output_run`` does not exist.

    Notes
    -----
    Constructing a `QuantumGraphBuilder` will run queries for existing datasets
    with empty data IDs (including but not limited to init inputs and outputs),
    in addition to resolving the given pipeline graph and testing for existence
    of the ``output`` run collection.

    The `build` method splits the pipeline graph into independent subgraphs,
    then calls the abstract method `process_subgraph` on each, to allow
    concrete implementations to populate the rough graph structure (the
    `~.quantum_graph_skeleton.QuantumGraphSkeleton` class), including searching
    for existing datasets.  The `build` method then:

    - assembles `lsst.daf.butler.Quantum` instances from all data IDs in the
      skeleton;
    - looks for existing outputs found in ``skip_existing_in`` to see if any
      quanta should be skipped;
    - calls `PipelineTaskConnections.adjustQuantum` on all quanta, adjusting
      downstream quanta appropriately when preliminary predicted outputs are
      rejected (pruning nodes that will not have the inputs they need to run);
    - attaches datastore records and registry dataset types to the graph.

    In addition to implementing `process_subgraph`, derived classes are
    generally expected to add new construction keyword-only arguments to
    control the data IDs of the quantum graph, while forwarding all of the
    arguments defined in the base class to `super`.
    """

    def __init__(
        self,
        pipeline_graph: PipelineGraph,
        butler: Butler,
        *,
        input_collections: Sequence[str] | None = None,
        output_run: str | None = None,
        skip_existing_in: Sequence[str] = (),
        retained_dataset_types: Sequence[str] | None = None,
        prune_unanchored_quanta: tuple[str, str] | None = None,
        clobber: bool = False,
    ):
        self.log = getLogger(__name__)
        self.metadata = TaskMetadata()
        self._pipeline_graph = pipeline_graph
        if input_collections is None:
            input_collections = butler.collections.defaults
        if not input_collections:
            raise ValueError("No input collections provided.")
        self.input_collections = input_collections
        if output_run is None:
            output_run = butler.run
        if not output_run:
            raise ValueError("No output RUN collection provided.")
        self.butler = butler.clone(collections=input_collections)
        self.output_run = output_run
        self.skip_existing_in = skip_existing_in
        self._retained_dataset_type_patterns: list[str] | None = (
            list(retained_dataset_types) if retained_dataset_types is not None else None
        )
        if self._retained_dataset_type_patterns is not None and not skip_existing_in:
            raise ValueError("retained_dataset_types has no effect without skip_existing_in.")
        self.empty_data_id = DataCoordinate.make_empty(butler.dimensions)
        self.clobber = clobber
        self._prune_unanchored_quanta = prune_unanchored_quanta
        # See whether the output run already exists.
        self.output_run_exists = False
        try:
            if self.butler.registry.getCollectionType(self.output_run) is not CollectionType.RUN:
                raise RuntimeError(f"{self.output_run!r} is not a RUN collection.")
            self.output_run_exists = True
        except MissingCollectionError:
            # If the run doesn't exist we never need to clobber.  This is not
            # an error so you can run with clobber=True the first time you
            # attempt some processing as well as all subsequent times, instead
            # of forcing the user to make the first attempt different.
            self.clobber = False
        # We need to know whether the skip_existing_in collection sequence
        # starts with the output run collection, as an optimization to avoid
        # queries later.
        try:
            skip_existing_in_flat = self.butler.collections.query(self.skip_existing_in, flatten_chains=True)
        except MissingCollectionError:
            skip_existing_in_flat = []
        if not skip_existing_in_flat:
            self.skip_existing_in = []
        if self.skip_existing_in and self.output_run_exists:
            self.skip_existing_starts_with_output_run = self.output_run == skip_existing_in_flat[0]
        else:
            self.skip_existing_starts_with_output_run = False
        try:
            packages_storage_class = butler.get_dataset_type(acc.PACKAGES_INIT_OUTPUT_NAME).storageClass_name
        except MissingDatasetTypeError:
            packages_storage_class = acc.PACKAGES_INIT_OUTPUT_STORAGE_CLASS
        self._global_init_output_types = {
            acc.PACKAGES_INIT_OUTPUT_NAME: DatasetType(
                acc.PACKAGES_INIT_OUTPUT_NAME,
                self.universe.empty,
                packages_storage_class,
            )
        }
        with self.butler.registry.caching_context():
            self._pipeline_graph.resolve(self.butler.registry)
            self.empty_dimensions_datasets = self._find_empty_dimension_datasets()
            self.prerequisite_info = {
                task_node.label: PrerequisiteInfo(task_node, self._pipeline_graph)
                for task_node in pipeline_graph.tasks.values()
            }
        if self._prune_unanchored_quanta is not None:
            source_label, anchor_label = self._prune_unanchored_quanta
            if source_label not in self._pipeline_graph.tasks:
                self.log.warning(
                    "prune_unanchored_quanta source label %r is not present in the pipeline; "
                    "pruning will have no effect.",
                    source_label,
                )
            elif anchor_label not in self._pipeline_graph.tasks:
                self.log.warning(
                    "prune_unanchored_quanta anchor label %r is not present in the pipeline; "
                    "all %r quanta will be treated as unanchored and removed.",
                    anchor_label,
                    source_label,
                )

    log: LsstLogAdapter
    """Logger to use for all quantum-graph generation messages.

    General and per-task status messages should be logged at `~logging.INFO`
    level or higher, per-dataset-type status messages should be logged at
    `~lsst.utils.logging.VERBOSE` or higher, and per-data-ID status messages
    should be logged at `logging.DEBUG` or higher.
    """

    metadata: TaskMetadata
    """Metadata to store in the QuantumGraph.

    The `TaskMetadata` class is used here primarily in order to enable
    resource-usage collection with the `lsst.utils.timer.timeMethod` decorator.
    """

    butler: Butler
    """Client for the data repository.

    Should be read-only.
    """

    input_collections: Sequence[str]
    """Collections to search for overall-input datasets.
    """

    output_run: str
    """Output `~lsst.daf.butler.CollectionType.RUN` collection.
    """

    skip_existing_in: Sequence[str]
    """Collections to search for outputs that already exist for the purpose
    of skipping quanta that have already been run.
    """

    clobber: bool
    """Whether to raise if predicted outputs already exist in ``output_run``

    This never actually clobbers outputs; it just informs the graph generation
    algorithm whether execution will run with clobbering enabled.  This is
    always `False` if `output_run_exists` is `False`.
    """

    empty_data_id: DataCoordinate
    """An empty data ID in the data repository's dimension universe.
    """

    output_run_exists: bool
    """Whether the output run exists in the data repository already.
    """

    skip_existing_starts_with_output_run: bool
    """Whether the `skip_existing_in` sequence begins with `output_run`.

    If this is true, any dataset found in `output_run` can be used to
    short-circuit queries in `skip_existing_in`.
    """

    empty_dimensions_datasets: EmptyDimensionsDatasets
    """Struct holding datasets with empty dimensions that have already been
    found in the data repository.
    """

    prerequisite_info: Mapping[str, PrerequisiteInfo]
    """Helper objects for finding prerequisite inputs, organized by task label.

    Subclasses that find prerequisites should remove the
    covered `~prerequisite_helpers.PrerequisiteFinder` objects from this
    attribute.
    """

    @property
    def universe(self) -> DimensionUniverse:
        """Definitions of all data dimensions."""
        return self.butler.dimensions

    @final
    @timeMethod
    def build(
        self, metadata: Mapping[str, Any] | None = None, attach_datastore_records: bool = True
    ) -> QuantumGraph:
        """Build the quantum graph, returning an old `QuantumGraph` instance.

        Parameters
        ----------
        metadata : `~collections.abc.Mapping`, optional
            Flexible metadata to add to the quantum graph.
        attach_datastore_records : `bool`, optional
            Whether to include datastore records in the graph.  Required for
            `lsst.daf.butler.QuantumBackedButler` execution.

        Returns
        -------
        quantum_graph : `.QuantumGraph`
            DAG describing processing to be performed.

        Notes
        -----
        External code is expected to construct a `QuantumGraphBuilder` and then
        call this method exactly once.  See class documentation for details on
        what it does.
        """
        skeleton = self._build_skeleton(attach_datastore_records=attach_datastore_records)
        if metadata is None:
            metadata = {
                "input": list(self.input_collections),
                "output_run": self.output_run,
            }
        return self._construct_quantum_graph(skeleton, metadata)

    def finish(
        self,
        output: str | None = None,
        metadata: Mapping[str, Any] | None = None,
        attach_datastore_records: bool = True,
    ) -> PredictedQuantumGraphComponents:
        """Return quantum graph components that can be used to save or
        construct a `PredictedQuantumGraph` instance.

        Parameters
        ----------
        output : `str` or `None`, optional
            Output `~lsst.daf.butler.CollectionType.CHAINED` collection that
            combines the input and output collections.
        metadata : `~collections.abc.Mapping`, optional
            Mapping of JSON-friendly metadata.  Collection information, the
            current user, and the current timestamp are automatically
            included.
        attach_datastore_records : `bool`, optional
            Whether to include datastore records for overall inputs for
            `~lsst.daf.butler.QuantumBackedButler`.

        Returns
        -------
        components : `.quantum_graph.PredictedQuantumGraphComponents`
            Components that can be used to construct a graph object and/or save
            it to disk.
        """
        skeleton = self._build_skeleton(attach_datastore_records=attach_datastore_records)
        return self._construct_components(skeleton, output=output, metadata=metadata)

    def _build_skeleton(self, attach_datastore_records: bool = True) -> QuantumGraphSkeleton:
        """Build a complete skeleton for the quantum graph.

        Parameters
        ----------
        attach_datastore_records : `bool`, optional
            Whether to include datastore records in the graph.  Required for
            `lsst.daf.butler.QuantumBackedButler` execution.

        Returns
        -------
        quantum_graph_skeleton : `QuantumGraphSkeleton`
            DAG describing processing to be performed.
        """
        with self.butler.registry.caching_context():
            full_skeleton = QuantumGraphSkeleton(self._pipeline_graph.tasks)
            subgraphs = list(self._pipeline_graph.split_independent())
            for i, subgraph in enumerate(subgraphs):
                self.log.info(
                    "Processing pipeline subgraph %d of %d with %d task(s).",
                    i + 1,
                    len(subgraphs),
                    len(subgraph.tasks),
                )
                self.log.verbose("Subgraph tasks: [%s]", ", ".join(label for label in subgraph.tasks))
                subgraph_skeleton = self.process_subgraph(subgraph)
                full_skeleton.update(subgraph_skeleton)
            # Loop over tasks to apply skip-existing logic and add missing
            # prerequisites.  The pipeline graph must be topologically sorted,
            # so a quantum is only processed after any quantum that provides
            # its inputs has been processed.
            skipped_quanta: dict[str, list[QuantumKey]] = {}
            retained_types = self._expand_retained_patterns(self._retained_dataset_type_patterns)
            # retained_types is None when all types are retained or option
            # absent: no ancestor unskipping is needed.
            if retained_types is None:
                for task_node in self._pipeline_graph.tasks.values():
                    skipped_quanta[task_node.label] = self._resolve_task_quanta(task_node, full_skeleton)
            else:
                skip_decisions: dict[QuantumKey, bool] = {}
                # Compute initial skip decisions without mutating the skeleton.
                for task_node in self._pipeline_graph.tasks.values():
                    for quantum_key in full_skeleton.get_quanta(task_node.label):
                        skip_decisions[quantum_key] = self._compute_skip_decision(
                            task_node, quantum_key, full_skeleton
                        )
                # Unskip ancestor quanta whose outputs are not retained.
                n_unskipped = self._unskip_ancestors(full_skeleton, skip_decisions, retained_types)
                if n_unskipped:
                    self.log.info(
                        "Forcing %s to rerun (output not retained).",
                        _quantum_or_quanta(n_unskipped),
                    )
                # Apply decisions.
                for task_node in self._pipeline_graph.tasks.values():
                    skipped_quanta[task_node.label] = self._resolve_task_quanta(
                        task_node, full_skeleton, skip_decisions=skip_decisions
                    )
            # Add any dimension records not handled by the subclass, and
            # aggregate any that were added directly to data IDs.
            full_skeleton.attach_dimension_records(self.butler, self._pipeline_graph.get_all_dimensions())
            # Loop over tasks again to run the adjust hooks.
            for task_node in self._pipeline_graph.tasks.values():
                self._adjust_task_quanta(task_node, full_skeleton, skipped_quanta[task_node.label])
            # Add global init-outputs to the skeleton.
            for dataset_type in self._global_init_output_types.values():
                dataset_key = full_skeleton.add_dataset_node(
                    dataset_type.name, self.empty_data_id, is_global_init_output=True
                )
                ref = self.empty_dimensions_datasets.outputs_in_the_way.get(dataset_key)
                if ref is None:
                    ref = DatasetRef(dataset_type, self.empty_data_id, run=self.output_run)
                full_skeleton.set_dataset_ref(ref, dataset_key)
            # Remove dataset nodes with no edges that are not global init
            # outputs, which are generally overall-inputs whose original quanta
            # end up skipped or with no work to do (we can't remove these along
            # with the quanta because no quantum knows if its the only
            # consumer).
            full_skeleton.remove_orphan_datasets()
            if self._prune_unanchored_quanta is not None:
                source_label, anchor_label = self._prune_unanchored_quanta
                n_source, n_downstream = full_skeleton.remove_unanchored_quanta(source_label, anchor_label)
                if n_source:
                    self.log.info(
                        "Pruned %d unanchored %r quanta and %d downstream quanta (%d total) based on %r.",
                        n_source,
                        source_label,
                        n_downstream,
                        n_source + n_downstream,
                        anchor_label,
                    )
                    full_skeleton.remove_orphan_datasets()
            if attach_datastore_records:
                self._attach_datastore_records(full_skeleton)
        return full_skeleton

    @abstractmethod
    def process_subgraph(self, subgraph: PipelineGraph) -> QuantumGraphSkeleton:
        """Build the rough structure for an independent subset of the
        `.QuantumGraph` and query for relevant existing datasets.

        Parameters
        ----------
        subgraph : `.pipeline_graph.PipelineGraph`
            Subset of the pipeline graph that should be processed by this call.
            This is always resolved and topologically sorted.  It should not be
            modified.

        Returns
        -------
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Class representing an initial quantum graph. See
            `.quantum_graph_skeleton.QuantumGraphSkeleton` docs for details.
            After this is returned, the object may be modified in-place in
            unspecified ways.

        Notes
        -----
        The `.quantum_graph_skeleton.QuantumGraphSkeleton` should associate
        `lsst.daf.butler.DatasetRef` objects with nodes for existing datasets.
        In particular:

        - `.quantum_graph_skeleton.QuantumGraphSkeleton.set_dataset_ref` must
          be used to associate existing datasets with all overall-input dataset
          nodes in the skeleton by querying `input_collections`.  This includes
          all standard input nodes and any prerequisite nodes added by the
          method (prerequisite nodes may also be left out entirely, as the base
          class can add them later, albeit possibly less efficiently).
        - `.quantum_graph_skeleton.QuantumGraphSkeleton.set_output_for_skip`
          must be used to associate existing datasets with output dataset nodes
          by querying `skip_existing_in`.
        - `.quantum_graph_skeleton.QuantumGraphSkeleton.add_output_in_the_way`
          must be used to associated existing outputs with output dataset nodes
          by querying `output_run` if `output_run_exists` is `True`. Note that
          the presence of such datasets is not automatically an error, even if
          `clobber` is `False`, as these may be quanta that will be skipped.

        `lsst.daf.butler.DatasetRef` objects for existing datasets with empty
        data IDs in all of the above categories may be found in the
        `empty_dimensions_datasets` attribute, as these are queried for prior
        to this call by the base class, but associating them with graph nodes
        is still this method's responsibility.

        Dataset types should never be components and should always use the
        "common" storage class definition in `pipeline_graph.DatasetTypeNode`
        (which is the data repository definition when the dataset type is
        registered).
        """
        raise NotImplementedError()

    @final
    @timeMethod
    def _resolve_task_quanta(
        self,
        task_node: TaskNode,
        skeleton: QuantumGraphSkeleton,
        skip_decisions: dict[QuantumKey, bool] | None = None,
    ) -> list[QuantumKey]:
        """Process the quanta for one task in a skeleton graph to skip those
        that have already completed and add missing prerequisite inputs.

        Parameters
        ----------
        task_node : `pipeline_graph.TaskNode`
            Node for this task in the pipeline graph.
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph, to be modified in-place.
        skip_decisions : `dict` [ `QuantumKey`, `bool` ] or `None`, optional
            Pre-computed per-quantum skip decisions.  When provided, the
            decisions are applied directly.

        Returns
        -------
        skipped_quanta : `list` [ `.quantum_skeleton_graph.QuantumKey` ]
            Keys of quanta that were already skipped because their metadata
            already exists in a ``skip_existing_in`` collections.

        Notes
        -----
        This method modifies ``skeleton`` in-place in several ways:

        - It associates a `lsst.daf.butler.DatasetRef` with all output datasets
          and drops input dataset nodes that do not have a
          `lsst.daf.butler.DatasetRef` already.  This ensures producing and
          consuming tasks start from the same `lsst.daf.butler.DatasetRef`.
        - It removes quantum nodes that are to be skipped because their outputs
          already exist in `skip_existing_in`.  It also marks their outputs
          as no longer in the way.
        - It adds prerequisite dataset nodes and edges that connect them to the
          quanta that consume them.
        """
        # Extract the helper object for the prerequisite inputs of this task,
        # and tell it to prepare to construct skypix bounds and timespans for
        # each quantum (these will automatically do nothing if nothing needs
        # those bounds).
        task_prerequisite_info = self.prerequisite_info[task_node.label]
        task_prerequisite_info.update_bounds()
        # Loop over all quanta for this task, remembering the ones we've
        # gotten rid of.
        skipped_quanta = []
        for quantum_key in skeleton.get_quanta(task_node.label):
            if skip_decisions is not None:
                if skip_decisions.get(quantum_key, False):
                    self._apply_skip_decision(task_node, quantum_key, skeleton)
                    skipped_quanta.append(quantum_key)
                    continue
            elif self._compute_skip_decision(task_node, quantum_key, skeleton):
                self._apply_skip_decision(task_node, quantum_key, skeleton)
                skipped_quanta.append(quantum_key)
                continue
            quantum_data_id = skeleton[quantum_key]["data_id"]
            skypix_bounds_builder = task_prerequisite_info.bounds.make_skypix_bounds_builder(quantum_data_id)
            timespan_builder = task_prerequisite_info.bounds.make_timespan_builder(quantum_data_id)
            self._update_quantum_for_adjust(
                quantum_key,
                skeleton,
                task_prerequisite_info,
                skypix_bounds_builder,
                timespan_builder,
            )
        for skipped_quantum in skipped_quanta:
            skeleton.remove_quantum_node(skipped_quantum, remove_outputs=False)
        return skipped_quanta

    @final
    @timeMethod
    def _adjust_task_quanta(
        self, task_node: TaskNode, skeleton: QuantumGraphSkeleton, skipped_quanta: list[QuantumKey]
    ) -> None:
        """Process the quanta for one task in a skeleton graph by calling the
        ``adjust_all_quanta`` and ``adjustQuantum`` hooks.

        Parameters
        ----------
        task_node : `pipeline_graph.TaskNode`
            Node for this task in the pipeline graph.
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph, to be modified in-place.
        skipped_quanta : `list` [ `.quantum_skeleton_graph.QuantumKey` ]
            Keys of quanta that were already skipped because their metadata
            already exists in a ``skip_existing_in`` collections.

        Notes
        -----
        This method modifies ``skeleton`` in-place in several ways:

        - It adds "inputs", "outputs", and "init_inputs" attributes to the
          quantum nodes, holding the same `NamedValueMapping` objects needed to
          construct an actual `Quantum` instances.
        - It removes quantum nodes whose
          `~PipelineTaskConnections.adjustQuantum` calls raise `NoWorkFound` or
          predict no outputs;
        - It removes the nodes of output datasets that are "adjusted away".
        - It removes the edges of input datasets that are "adjusted away".

        The difference between how adjusted inputs and outputs are handled
        reflects the fact that many quanta can share the same input, but only
        one produces each output.  This can lead to the graph having
        superfluous isolated nodes after processing is complete, but these
        should only be removed after all the quanta from all tasks have been
        processed.
        """
        # Give the task a chance to adjust all quanta together.  This
        # operates directly on the skeleton (via a the 'adjuster', which
        # is just an interface adapter).
        adjuster = QuantaAdjuster(task_node.label, self._pipeline_graph, skeleton, self.butler)
        task_node.get_connections().adjust_all_quanta(adjuster)
        # Loop over all quanta again, remembering those we get rid of in other
        # ways.
        no_work_quanta = []
        for quantum_key in skeleton.get_quanta(task_node.label):
            adjusted_outputs = self._adapt_quantum_outputs(task_node, quantum_key, skeleton)
            adjusted_inputs = self._adapt_quantum_inputs(task_node, quantum_key, skeleton)
            # Give the task's Connections class an opportunity to remove
            # some inputs, or complain if they are unacceptable.  This will
            # raise if one of the check conditions is not met, which is the
            # intended behavior.
            helper = AdjustQuantumHelper(inputs=adjusted_inputs, outputs=adjusted_outputs)
            quantum_data_id = skeleton[quantum_key]["data_id"]
            try:
                helper.adjust_in_place(task_node.get_connections(), task_node.label, quantum_data_id)
            except NoWorkFound as err:
                # Do not generate this quantum; it would not produce any
                # outputs.  Remove it and all of the outputs it might have
                # produced from the skeleton.
                try:
                    _, connection_name, _ = err.args
                    details = f"not enough datasets for connection {connection_name}."
                except ValueError:
                    details = str(err)
                self.log.debug(
                    "No work found for quantum %s of task %s: %s",
                    quantum_key.data_id_values,
                    quantum_key.task_label,
                    details,
                )
                no_work_quanta.append(quantum_key)
                continue
            if helper.outputs_adjusted:
                if not any(adjusted_refs for adjusted_refs in helper.outputs.values()):
                    # No outputs also means we don't generate this quantum.
                    self.log.debug(
                        "No outputs predicted for quantum %s of task %s.",
                        quantum_key.data_id_values,
                        quantum_key.task_label,
                    )
                    no_work_quanta.append(quantum_key)
                    continue
                # Remove output nodes that were not retained by
                # adjustQuantum.
                skeleton.remove_dataset_nodes(
                    self._find_removed(skeleton.iter_outputs_of(quantum_key), helper.outputs)
                )
            if helper.inputs_adjusted:
                if not any(bool(adjusted_refs) for adjusted_refs in helper.inputs.values()):
                    raise QuantumGraphBuilderError(
                        f"adjustQuantum implementation for {task_node.label}@{quantum_key.data_id_values} "
                        "returned outputs but no inputs."
                    )
                # Remove input dataset edges that were not retained by
                # adjustQuantum.  We can't remove the input dataset nodes
                # because some other quantum might still want them.
                skeleton.remove_input_edges(
                    quantum_key, self._find_removed(skeleton.iter_inputs_of(quantum_key), helper.inputs)
                )
            # Save the adjusted inputs and outputs to the quantum node's
            # state so we don't have to regenerate those data structures
            # from the graph.
            skeleton[quantum_key]["inputs"] = helper.inputs
            skeleton[quantum_key]["outputs"] = helper.outputs
        for no_work_quantum in no_work_quanta:
            skeleton.remove_quantum_node(no_work_quantum, remove_outputs=True)
        remaining_quanta = skeleton.get_quanta(task_node.label)
        self._resolve_task_init(task_node, skeleton, bool(skipped_quanta))
        message_terms = []
        if no_work_quanta:
            message_terms.append(f"{len(no_work_quanta)} had no work to do")
        if skipped_quanta:
            message_terms.append(f"{len(skipped_quanta)} previously succeeded and skipped")
        if adjuster.n_removed:
            message_terms.append(f"{adjuster.n_removed} removed by adjust_all_quanta")
        message_parenthetical = f" ({', '.join(message_terms)})" if message_terms else ""
        if remaining_quanta:
            self.log.info(
                "Generated %s for task %s%s.",
                _quantum_or_quanta(len(remaining_quanta)),
                task_node.label,
                message_parenthetical,
            )
        else:
            self.log.info(
                "Dropping task %s because no quanta remain%s.", task_node.label, message_parenthetical
            )
            skeleton.remove_task(task_node.label)
        if len(no_work_quanta) > len(remaining_quanta):
            only_overall_inputs = self._get_task_inputs_if_overall_only(task_node)
            self.log.warning(
                "More than half of %s quanta had no work to do given available inputs.\n"
                "A query constraint on one of %s may yield a much faster build.",
                task_node.label,
                only_overall_inputs,
            )

    def _get_task_inputs_if_overall_only(self, task_node: TaskNode) -> list[str] | None:
        """If the given task consumes only overall-inputs, return their names.
        Otherwise return `None`.
        """
        result: list[str] = []
        for read_edge in task_node.inputs.values():
            if self._pipeline_graph.producer_of(read_edge.parent_dataset_type_name) is None:
                result.append(read_edge.parent_dataset_type_name)
            else:
                return None
        return result

    def _compute_skip_decision(
        self, task_node: TaskNode, quantum_key: QuantumKey, skeleton: QuantumGraphSkeleton
    ) -> bool:
        """Identify if a quantum should be skipped because its
        metadata dataset already exists.

        Parameters
        ----------
        task_node : `pipeline_graph.TaskNode`
            Node for this task in the pipeline graph.
        quantum_key : `QuantumKey`
            Identifier for this quantum in the graph.
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph (not modified).

        Returns
        -------
        skip : `bool`
            `True` if the quantum's metadata exists in ``skip_existing_in`` and
            should be skipped.
        """
        metadata_dataset_key = DatasetKey(
            task_node.metadata_output.parent_dataset_type_name, quantum_key.data_id_values
        )
        return bool(skeleton.get_output_for_skip(metadata_dataset_key))

    def _apply_skip_decision(
        self, task_node: TaskNode, quantum_key: QuantumKey, skeleton: QuantumGraphSkeleton
    ) -> None:
        """Update the skeleton for a quantum that has been decided to skip.

        Parameters
        ----------
        task_node : `pipeline_graph.TaskNode`
            Node for this task in the pipeline graph.
        quantum_key : `QuantumKey`
            Identifier for this quantum in the graph.
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph, to be modified in-place.

        Notes
        -----
        The metadata dataset for this quantum exists in the
        `skip_existing_in` collections and the quantum will be skipped. This
        causes the quantum node to be removed from the graph.  Dataset nodes
        that were previously the outputs of this quantum will be associated
        with `lsst.daf.butler.DatasetRef` objects that were found in
        ``skip_existing_in``, or will be removed if there is no such dataset
        there.  Any output dataset in `output_run` will be removed from the
        "output in the way" category.
        """
        # This quantum's metadata is already present in the
        # skip_existing_in collections; we'll skip it.  But the presence of
        # the metadata dataset doesn't guarantee that all of the other
        # outputs we predicted are present; we have to check.
        for output_dataset_key in list(skeleton.iter_outputs_of(quantum_key)):
            # If this dataset was "in the way" (i.e. already in the
            # output run), it isn't anymore.
            skeleton.discard_output_in_the_way(output_dataset_key)
            if (output_ref := skeleton.get_output_for_skip(output_dataset_key)) is not None:
                # Populate the skeleton graph's node attributes
                # with the existing DatasetRef, just like a
                # predicted output of a non-skipped quantum.
                skeleton.set_dataset_ref(output_ref, output_dataset_key)
            else:
                # Remove this dataset from the skeleton graph,
                # because the quantum that would have produced it
                # is being skipped and it doesn't already exist.
                skeleton.remove_dataset_nodes([output_dataset_key])
        # Removing the quantum node from the graph will happen outside this
        # function.

    def _expand_retained_patterns(self, patterns: list[str] | None) -> frozenset[str] | None:
        """Expand wildcard patterns into a concrete set of retained dataset
        type names.

        Parameters
        ----------
        patterns : `list` [ `str` ] or `None`
            Dataset type names or glob-style wildcard patterns, or `None` if
            the option was not provided.

        Returns
        -------
        retained_types : `frozenset` [ `str` ] or `None`
            Concrete set of retained dataset type names, or `None` if no
            ancestor unskipping is needed (option absent, empty list, or
            patterns match everything).

        """
        if patterns is None:
            return None
        regexes = globToRegex(patterns)
        if regexes is ...:
            # globToRegex returns Ellipsis when patterns match everything,
            # e.g. the list is empty or "*".  Treat as "retain everything".
            return None
        all_names = set(self._pipeline_graph.dataset_types)
        result: set[str] = set()
        for original, expression in zip(patterns, regexes):
            if isinstance(expression, str):
                if expression not in all_names:
                    self.log.warning("Retained dataset type %r not found in the pipeline.", expression)
                result.add(expression)
            else:
                matches = {n for n in all_names if expression.search(n)}
                if not matches:
                    self.log.warning(
                        "Retained dataset type pattern %r matches no dataset types.",
                        original,
                    )
                result.update(matches)
        return frozenset(result)

    def _unskip_ancestors(
        self,
        skeleton: QuantumGraphSkeleton,
        skip_decisions: dict[QuantumKey, bool],
        retained: frozenset[str],
    ) -> int:
        """Unskip ancestor quanta whose outputs are not retained.

        Parameters
        ----------
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph (not modified).
        skip_decisions : `dict` [ `QuantumKey`, `bool` ]
            Per-quantum skip decisions, modified in-place.
        retained : `frozenset` [ `str` ]
            Dataset type names that should be present in
            ``skip_existing_in`` when their producing task has been skipped.
            Types not in this set are treated as not retained.

        Returns
        -------
        n_unskipped : `int`
            Number of quanta unskipped by backward propagation.

        Notes
        -----
        Seeds the breadth-first search with every initially must-run quantum.
        For each must-run quantum, walks non-prerequisite input edges whose
        dataset type is not retained.  If the producer of such a dataset is
        currently marked skip, it is unskipped and enqueued so that its own
        inputs are examined in turn.  Each quantum is visited at most once.
        """
        queue: collections.deque[QuantumKey] = collections.deque(
            qk for qk, skip in skip_decisions.items() if not skip
        )
        visited: set[QuantumKey] = set(queue)
        n_unskipped = 0
        while queue:
            qk = queue.popleft()
            for input_key in skeleton.iter_inputs_of(qk):
                if input_key.is_prerequisite:
                    continue
                if input_key.parent_dataset_type_name in retained:
                    continue
                producer_key = skeleton.get_producer_of(input_key)
                if not isinstance(producer_key, QuantumKey):
                    continue
                if producer_key in visited:
                    continue
                visited.add(producer_key)
                if skip_decisions.get(producer_key, False):
                    skip_decisions[producer_key] = False
                    queue.append(producer_key)
                    n_unskipped += 1
        return n_unskipped

    @final
    def _update_quantum_for_adjust(
        self,
        quantum_key: QuantumKey,
        skeleton: QuantumGraphSkeleton,
        task_prerequisite_info: PrerequisiteInfo,
        skypix_bounds_builder: SkyPixBoundsBuilder,
        timespan_builder: TimespanBuilder,
    ) -> None:
        """Update the quantum node in the skeleton by finding remaining
        prerequisite inputs and dropping regular inputs that we now know will
        not be produced.

        Parameters
        ----------
        quantum_key : `QuantumKey`
            Identifier for this quantum in the graph.
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph, to be modified in-place.
        task_prerequisite_info : `~prerequisite_helpers.PrerequisiteInfo`
            Information about the prerequisite inputs to this task.
        skypix_bounds_builder : `~prerequisite_helpers.SkyPixBoundsBuilder`
            An object that accumulates the appropriate spatial bounds for a
            quantum.
        timespan_builder : `~prerequisite_helpers.TimespanBuilder`
            An object that accumulates the appropriate timespan for a quantum.

        Notes
        -----
        This first looks for outputs already present in the `output_run` (i.e.
        "in the way" in the skeleton); if it finds something and `clobber` is
        `True`, it uses that ref (it's not ideal that both the original dataset
        and its replacement will have the same UUID, but we don't have space in
        the quantum graph for two UUIDs, and we need the datastore records of
        the original there).  If `clobber` is `False`, `RuntimeError` is
        raised.  If there is no output already present, a new one with a random
        UUID is generated.  In all cases the dataset node in the skeleton is
        associated with a `lsst.daf.butler.DatasetRef`.
        """
        dataset_key: DatasetKey | PrerequisiteDatasetKey
        for dataset_key in skeleton.iter_outputs_of(quantum_key):
            dataset_data_id = skeleton.get_data_id(dataset_key)
            dataset_type_node = self._pipeline_graph.dataset_types[dataset_key.parent_dataset_type_name]
            if (ref := skeleton.get_output_in_the_way(dataset_key)) is None:
                ref = DatasetRef(dataset_type_node.dataset_type, dataset_data_id, run=self.output_run)
            elif not self.clobber:
                # We intentionally raise here, before running adjustQuantum,
                # because it'd be weird if we left an old potential output of a
                # task sitting there in the output collection, just because the
                # task happened to not actually produce it.
                raise OutputExistsError(
                    f"Potential output dataset {ref} already exists in the output run "
                    f"{self.output_run}, but clobbering outputs was not expected to be necessary."
                )
            skypix_bounds_builder.handle_dataset(dataset_key.parent_dataset_type_name, dataset_data_id)
            timespan_builder.handle_dataset(dataset_key.parent_dataset_type_name, dataset_data_id)
            skeleton.set_dataset_ref(ref, dataset_key)
        quantum_data_id = skeleton.get_data_id(quantum_key)
        # Process inputs already present in the skeleton - this should include
        # all regular inputs (including intermediates) and may include some
        # prerequisites.
        for dataset_key in list(skeleton.iter_inputs_of(quantum_key)):
            if (ref := skeleton.get_dataset_ref(dataset_key)) is None:
                # If the dataset ref hasn't been set either as an existing
                # input or as an output of an already-processed upstream
                # quantum, it's not going to be produced; remove it.
                skeleton.remove_dataset_nodes([dataset_key])
                continue
            skypix_bounds_builder.handle_dataset(dataset_key.parent_dataset_type_name, ref.dataId)
            timespan_builder.handle_dataset(dataset_key.parent_dataset_type_name, ref.dataId)
        # Query for any prerequisites not handled by process_subgraph.  Note
        # that these were not already in the skeleton graph, so we add them
        # now.
        skypix_bounds = skypix_bounds_builder.finish()
        timespan = timespan_builder.finish()
        for finder in task_prerequisite_info.finders.values():
            dataset_keys = []
            for ref in finder.find(
                self.butler, self.input_collections, quantum_data_id, skypix_bounds, timespan
            ):
                dataset_key = skeleton.add_prerequisite_node(ref)
                dataset_keys.append(dataset_key)
            skeleton.add_input_edges(quantum_key, dataset_keys)

    @final
    def _adapt_quantum_outputs(
        self,
        task_node: TaskNode,
        quantum_key: QuantumKey,
        skeleton: QuantumGraphSkeleton,
    ) -> NamedKeyDict[DatasetType, list[DatasetRef]]:
        """Adapt outputs for a preliminary quantum and put them into the form
        used by `~lsst.daf.butler.Quantum` and
        `~PipelineTaskConnections.adjustQuantum`.

        Parameters
        ----------
        task_node : `pipeline_graph.TaskNode`
            Node for this task in the pipeline graph.
        quantum_key : `QuantumKey`
            Identifier for this quantum in the graph.
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph, to be modified in-place.

        Returns
        -------
        outputs : `~lsst.daf.butler.NamedKeyDict` [ \
                `~lsst.daf.butler.DatasetType`, `list` [ \
                `~lsst.daf.butler.DatasetRef` ] ]
            All outputs to the task, using the storage class and components
            defined by the task's own connections.
        """
        outputs_by_type: dict[str, list[DatasetRef]] = {}
        dataset_key: DatasetKey
        for dataset_key in skeleton.iter_outputs_of(quantum_key):
            ref = skeleton.get_dataset_ref(dataset_key)
            assert ref is not None, "Should have been added (or the node removed) in a previous pass."
            outputs_by_type.setdefault(dataset_key.parent_dataset_type_name, []).append(ref)
        adapted_outputs: NamedKeyDict[DatasetType, list[DatasetRef]] = NamedKeyDict()
        for write_edge in task_node.iter_all_outputs():
            dataset_type_node = self._pipeline_graph.dataset_types[write_edge.parent_dataset_type_name]
            edge_dataset_type = write_edge.adapt_dataset_type(dataset_type_node.dataset_type)
            adapted_outputs[edge_dataset_type] = [
                write_edge.adapt_dataset_ref(ref)
                for ref in sorted(outputs_by_type.get(write_edge.parent_dataset_type_name, []))
            ]
        return adapted_outputs

    @final
    def _adapt_quantum_inputs(
        self,
        task_node: TaskNode,
        quantum_key: QuantumKey,
        skeleton: QuantumGraphSkeleton,
    ) -> NamedKeyDict[DatasetType, list[DatasetRef]]:
        """Adapt input datasets for a preliminary quantum into the form used by
        `~lsst.daf.butler.Quantum` and
        `~PipelineTaskConnections.adjustQuantum`.

        Parameters
        ----------
        task_node : `pipeline_graph.TaskNode`
            Node for this task in the pipeline graph.
        quantum_key : `QuantumKey`
            Identifier for this quantum in the graph.
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph, to be modified in-place.

        Returns
        -------
        inputs : `~lsst.daf.butler.NamedKeyDict` [ \
                `~lsst.daf.butler.DatasetType`, `list` [ \
                `~lsst.daf.butler.DatasetRef` ] ]
            All regular and prerequisite inputs to the task, using the storage
            class and components defined by the task's own connections.

        Notes
        -----
        This method trims input dataset nodes that are not already associated
        with a `lsst.daf.butler.DatasetRef`, and queries for prerequisite input
        nodes that do not exist.
        """
        inputs_by_type: dict[str, set[DatasetRef]] = {}
        dataset_key: DatasetKey | PrerequisiteDatasetKey
        for dataset_key in list(skeleton.iter_inputs_of(quantum_key)):
            ref = skeleton.get_dataset_ref(dataset_key)
            assert ref is not None, "Should have been added (or the node removed) in a previous pass."
            inputs_by_type.setdefault(dataset_key.parent_dataset_type_name, set()).add(ref)
        adapted_inputs: NamedKeyDict[DatasetType, list[DatasetRef]] = NamedKeyDict()
        for read_edge in task_node.iter_all_inputs():
            dataset_type_node = self._pipeline_graph.dataset_types[read_edge.parent_dataset_type_name]
            edge_dataset_type = read_edge.adapt_dataset_type(dataset_type_node.dataset_type)
            if (current_dataset_type := adapted_inputs.keys().get(edge_dataset_type.name)) is None:
                adapted_inputs[edge_dataset_type] = [
                    read_edge.adapt_dataset_ref(ref)
                    for ref in sorted(inputs_by_type.get(read_edge.parent_dataset_type_name, frozenset()))
                ]
            elif current_dataset_type != edge_dataset_type:
                raise NotImplementedError(
                    f"Task {task_node.label!r} has {edge_dataset_type.name!r} as an input via "
                    "two different connections, with two different storage class overrides. "
                    "This is not yet supported due to limitations in the Quantum data structure."
                )
            # If neither the `if` nor the `elif` above match, it means
            # multiple input connections have exactly the same dataset
            # type, and hence nothing to do after the first one.
        return adapted_inputs

    @final
    def _resolve_task_init(
        self, task_node: TaskNode, skeleton: QuantumGraphSkeleton, has_skipped_quanta: bool
    ) -> None:
        """Add init-input and init-output dataset nodes and edges for a task to
        the skeleton.

        Parameters
        ----------
        task_node : `pipeline_graph.TaskNode`
            Pipeline graph description of the task.
        skeleton : `QuantumGraphSkeleton`
            In-progress quantum graph data structure to update in-place.
        has_skipped_quanta : `bool`
            Whether any of this task's quanta were skipped because they had
            already succeeded.
        """
        quanta = skeleton.get_quanta(task_node.label)
        task_init_key = TaskInitKey(task_node.label)
        if quanta:
            adapted_inputs: NamedKeyDict[DatasetType, DatasetRef] = NamedKeyDict()
            # Process init-inputs.
            input_keys: list[DatasetKey] = []
            for read_edge in task_node.init.iter_all_inputs():
                dataset_key = skeleton.add_dataset_node(
                    read_edge.parent_dataset_type_name, self.empty_data_id
                )
                skeleton.add_input_edge(task_init_key, dataset_key)
                if (ref := skeleton.get_dataset_ref(dataset_key)) is None:
                    try:
                        ref = self.empty_dimensions_datasets.inputs[dataset_key]
                    except KeyError:
                        raise InitInputMissingError(
                            f"Overall init-input dataset {read_edge.parent_dataset_type_name!r} "
                            f"needed by task {task_node.label!r} not found in input collection(s) "
                            f"{self.input_collections}."
                        ) from None
                    skeleton.set_dataset_ref(ref, dataset_key)
                for quantum_key in skeleton.get_quanta(task_node.label):
                    skeleton.add_input_edge(quantum_key, dataset_key)
                input_keys.append(dataset_key)
                adapted_ref = read_edge.adapt_dataset_ref(ref)
                adapted_inputs[adapted_ref.datasetType] = adapted_ref
            # Save the quantum-adapted init inputs to each quantum, and add
            # skeleton edges connecting the init inputs to each quantum.
            for quantum_key in skeleton.get_quanta(task_node.label):
                skeleton[quantum_key]["init_inputs"] = adapted_inputs
            # Process init-outputs.
            adapted_outputs: NamedKeyDict[DatasetType, DatasetRef] = NamedKeyDict()
            for write_edge in task_node.init.iter_all_outputs():
                dataset_key = skeleton.add_dataset_node(
                    write_edge.parent_dataset_type_name, self.empty_data_id
                )
                if (ref := self.empty_dimensions_datasets.outputs_in_the_way.get(dataset_key)) is None:
                    ref = DatasetRef(
                        self._pipeline_graph.dataset_types[write_edge.parent_dataset_type_name].dataset_type,
                        self.empty_data_id,
                        run=self.output_run,
                    )
                skeleton.set_dataset_ref(ref, dataset_key)
                skeleton.add_output_edge(task_init_key, dataset_key)
                adapted_ref = write_edge.adapt_dataset_ref(ref)
                adapted_outputs[adapted_ref.datasetType] = adapted_ref
            skeleton[task_init_key]["inputs"] = adapted_inputs
            skeleton[task_init_key]["outputs"] = adapted_outputs
        elif has_skipped_quanta:
            # No quanta remain for this task, but at least one quantum was
            # skipped because its outputs were present in the skip_existing_in
            # collections.  This means all init outputs should be present in
            # the skip_existing_in collections, too, and we need to put those
            # refs in the graph.
            for write_edge in task_node.init.iter_all_outputs():
                dataset_key = skeleton.add_dataset_node(
                    write_edge.parent_dataset_type_name, self.empty_data_id
                )
                if (ref := self.empty_dimensions_datasets.outputs_for_skip.get(dataset_key)) is None:
                    raise InitInputMissingError(
                        f"Init-output dataset {write_edge.parent_dataset_type_name!r} of skipped task "
                        f"{task_node.label!r} not found in skip-existing-in collection(s) "
                        f"{self.skip_existing_in}."
                    ) from None
                skeleton.set_dataset_ref(ref, dataset_key)
                # If this dataset was "in the way" (i.e. already in the output
                # run), it isn't anymore.
                skeleton.discard_output_in_the_way(dataset_key)
        # No quanta remain in this task, but none were skipped; this means
        # they all got pruned because of NoWorkFound conditions.  This
        # dooms all downstream quanta to the same fate, so we don't bother
        # doing anything with the task's init-outputs, since nothing is
        # going to consume them.

    @final
    @timeMethod
    def _find_empty_dimension_datasets(self) -> EmptyDimensionsDatasets:
        """Query for all dataset types with no dimensions, updating
        `empty_dimensions_datasets` in-place.

        This includes but is not limited to init inputs and init outputs.
        """
        inputs: dict[DatasetKey | PrerequisiteDatasetKey, DatasetRef] = {}
        outputs_for_skip: dict[DatasetKey, DatasetRef] = {}
        outputs_in_the_way: dict[DatasetKey, DatasetRef] = {}
        _, dataset_type_nodes = self._pipeline_graph.group_by_dimensions().get(self.universe.empty, ({}, {}))
        dataset_types = [node.dataset_type for node in dataset_type_nodes.values()]
        dataset_types.extend(self._global_init_output_types.values())
        for dataset_type in dataset_types:
            key = DatasetKey(dataset_type.name, self.empty_data_id.required_values)
            if (
                self._pipeline_graph.producer_of(dataset_type.name) is None
                and dataset_type.name not in self._global_init_output_types
            ):
                # Dataset type is an overall input; we always need to try to
                # find these.
                try:
                    ref = self.butler.find_dataset(dataset_type.name, collections=self.input_collections)
                except MissingDatasetTypeError:
                    ref = None
                if ref is not None:
                    inputs[key] = ref
            elif self.skip_existing_in:
                # Dataset type is an intermediate or output; need to find these
                # if only they're from previously executed quanta that we might
                # skip...
                try:
                    ref = self.butler.find_dataset(dataset_type.name, collections=self.skip_existing_in)
                except MissingDatasetTypeError:
                    ref = None
                if ref is not None:
                    outputs_for_skip[key] = ref
                    if ref.run == self.output_run:
                        outputs_in_the_way[key] = ref
            if self.output_run_exists and not self.skip_existing_starts_with_output_run:
                # ...or if they're in the way and would need to be clobbered
                # (and we haven't already found them in the previous block).
                try:
                    ref = self.butler.find_dataset(dataset_type.name, collections=[self.output_run])
                except MissingDatasetTypeError:
                    ref = None
                if ref is not None:
                    outputs_in_the_way[key] = ref
        return EmptyDimensionsDatasets(
            inputs=inputs, outputs_for_skip=outputs_for_skip, outputs_in_the_way=outputs_in_the_way
        )

    @final
    @timeMethod
    def _attach_datastore_records(self, skeleton: QuantumGraphSkeleton) -> None:
        """Add datastore records for all overall inputs to a preliminary
        quantum graph.

        Parameters
        ----------
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph to update in place.

        Notes
        -----
        On return, all quantum nodes in the skeleton graph will have a
        "datastore_records" attribute that is a mapping from datastore name
        to `lsst.daf.butler.DatastoreRecordData`, as used by
        `lsst.daf.butler.Quantum`.
        """
        self.log.info("Fetching and attaching datastore records for all overall inputs.")
        overall_inputs = skeleton.extract_overall_inputs()
        exported_records = self.butler._datastore.export_records(overall_inputs.values())
        for task_label in self._pipeline_graph.tasks:
            if not skeleton.has_task(task_label):
                continue
            self.log.verbose("Fetching and attaching datastore records for task %s.", task_label)
            task_init_key = skeleton.get_task_init_node(task_label)
            init_input_ids = {
                ref.id
                for dataset_key in skeleton.iter_inputs_of(task_init_key)
                if (ref := overall_inputs.get(dataset_key)) is not None
            }
            init_records = {}
            if init_input_ids:
                for datastore_name, records in exported_records.items():
                    matching_records = records.subset(init_input_ids)
                    if matching_records is not None:
                        init_records[datastore_name] = matching_records
            skeleton[task_init_key]["datastore_records"] = init_records
            for quantum_key in skeleton.get_quanta(task_label):
                quantum_records = {}
                input_ids = {
                    ref.id
                    for dataset_key in skeleton.iter_inputs_of(quantum_key)
                    if (ref := overall_inputs.get(dataset_key)) is not None
                }
                if input_ids:
                    for datastore_name, records in exported_records.items():
                        matching_records = records.subset(input_ids)
                        if matching_records is not None:
                            quantum_records[datastore_name] = matching_records
                skeleton[quantum_key]["datastore_records"] = quantum_records

    @final
    @timeMethod
    def _construct_quantum_graph(
        self, skeleton: QuantumGraphSkeleton, metadata: Mapping[str, Any]
    ) -> QuantumGraph:
        """Construct a `.QuantumGraph` object from the contents of a
        fully-processed `.quantum_graph_skeleton.QuantumGraphSkeleton`.

        Parameters
        ----------
        skeleton : `.quantum_graph_skeleton.QuantumGraphSkeleton`
            Preliminary quantum graph.  Must have "init_inputs", "inputs", and
            "outputs" attributes on all quantum nodes, as added by
            `_resolve_task_quanta`, as well as a "datastore_records" attribute
            as added by `_attach_datastore_records`.
        metadata : `~collections.abc.Mapping`
            Flexible metadata to add to the graph.

        Returns
        -------
        quantum_graph : `.QuantumGraph`
            DAG describing processing to be performed.
        """
        from .graph import QuantumGraph

        self.log.info("Transforming graph skeleton into a QuantumGraph instance.")
        quanta: dict[TaskDef, set[Quantum]] = {}
        init_inputs: dict[TaskDef, Iterable[DatasetRef]] = {}
        init_outputs: dict[TaskDef, Iterable[DatasetRef]] = {}
        for task_def in self._pipeline_graph._iter_task_defs():
            if not skeleton.has_task(task_def.label):
                continue
            self.log.verbose("Transforming graph skeleton nodes for task %s.", task_def.label)
            task_node = self._pipeline_graph.tasks[task_def.label]
            task_init_key = skeleton.get_task_init_node(task_def.label)
            task_init_state = skeleton[task_init_key]
            init_datastore_records: dict[str, DatastoreRecordData] = task_init_state.get(
                "datastore_records", {}
            )
            init_inputs[task_def] = task_init_state["inputs"].values()
            init_outputs[task_def] = task_init_state["outputs"].values()
            quanta_for_task: set[Quantum] = set()
            for quantum_key in skeleton.get_quanta(task_node.label):
                quantum_state = skeleton[quantum_key]
                quantum_datastore_records: dict[str, DatastoreRecordData] = quantum_state.get(
                    "datastore_records", {}
                )
                quanta_for_task.add(
                    Quantum(
                        taskName=task_node.task_class_name,
                        taskClass=task_node.task_class,
                        dataId=quantum_state["data_id"],
                        initInputs=quantum_state["init_inputs"],
                        inputs=quantum_state["inputs"],
                        outputs=quantum_state["outputs"],
                        datastore_records=DatastoreRecordData.merge_mappings(
                            quantum_datastore_records, init_datastore_records
                        ),
                    )
                )
            quanta[task_def] = quanta_for_task

        registry_dataset_types: list[DatasetType] = [
            node.dataset_type for node in self._pipeline_graph.dataset_types.values()
        ]

        all_metadata = self.metadata.to_dict()
        all_metadata.update(metadata)
        global_init_outputs: list[DatasetRef] = []
        for dataset_key in skeleton.global_init_outputs:
            ref = skeleton.get_dataset_ref(dataset_key)
            assert ref is not None, "Global init input refs should be resolved already."
            global_init_outputs.append(ref)
        self.log.verbose("Invoking QuantumGraph class constructor.")
        result = QuantumGraph(
            quanta,
            metadata=all_metadata,
            universe=self.universe,
            initInputs=init_inputs,
            initOutputs=init_outputs,
            globalInitOutputs=global_init_outputs,
            registryDatasetTypes=registry_dataset_types,
        )
        self.log.info("Graph build complete.")
        return result

    @final
    @timeMethod
    def _construct_components(
        self,
        skeleton: QuantumGraphSkeleton,
        output: str | None,
        metadata: Mapping[str, Any] | None,
    ) -> PredictedQuantumGraphComponents:
        """Return quantum graph components from a completed skeleton.

        Parameters
        ----------
        skeleton : `quantum_graph_skeleton.QuantumGraphSkeleton`
            Temporary data structure used by the builder to represent the
            graph.
        output : `str` or `None`, optional
            Output `~lsst.daf.butler.CollectionType.CHAINED` collection that
            combines the input and output collections.
        metadata : `~collections.abc.Mapping`, optional
            Mapping of JSON-friendly metadata.  Collection information, the
            current user, and the current timestamp are automatically
            included.

        Returns
        -------
        components : `.quantum_graph.PredictedQuantumGraphComponents`
            Components that can be used to construct a graph object and/or save
            it to disk.
        """
        from .quantum_graph import (
            PredictedDatasetModel,
            PredictedQuantumDatasetsModel,
            PredictedQuantumGraphComponents,
        )

        self.log.info("Transforming graph skeleton into PredictedQuantumGraph components.")
        components = PredictedQuantumGraphComponents(pipeline_graph=self._pipeline_graph)
        components.header.inputs = list(self.input_collections)
        components.header.output_run = self.output_run
        components.header.output = output
        if metadata is not None:
            components.header.metadata.update(metadata)
        components.dimension_data = DimensionDataAttacher(
            records=skeleton.get_dimension_data(),
            dimensions=self._pipeline_graph.get_all_dimensions(),
        )
        components.init_quanta.root = [
            PredictedQuantumDatasetsModel.model_construct(
                quantum_id=generate_uuidv7(),
                task_label="",
                outputs={
                    dataset_key.parent_dataset_type_name: [
                        PredictedDatasetModel.from_dataset_ref(
                            cast(DatasetRef, skeleton.get_dataset_ref(dataset_key))
                        )
                    ]
                    for dataset_key in skeleton.global_init_outputs
                },
            )
        ]
        for task_node in self._pipeline_graph.tasks.values():
            if not skeleton.has_task(task_node.label):
                continue
            self.log.verbose("Transforming graph skeleton nodes for task %s.", task_node.label)
            task_init_key = TaskInitKey(task_node.label)
            init_quantum_datasets = PredictedQuantumDatasetsModel.model_construct(
                quantum_id=generate_uuidv7(),
                task_label=task_node.label,
                inputs=self._make_predicted_datasets(
                    skeleton,
                    task_node.init.iter_all_inputs(),
                    skeleton.iter_inputs_of(task_init_key),
                ),
                outputs=self._make_predicted_datasets(
                    skeleton,
                    task_node.init.iter_all_outputs(),
                    skeleton.iter_outputs_of(task_init_key),
                ),
                datastore_records={
                    datastore_name: records.to_simple()
                    for datastore_name, records in skeleton[task_init_key]
                    .get("datastore_records", {})
                    .items()
                },
            )
            components.init_quanta.root.append(init_quantum_datasets)
            for quantum_key in skeleton.get_quanta(task_node.label):
                quantum_datasets = PredictedQuantumDatasetsModel.model_construct(
                    quantum_id=generate_uuidv7(),
                    task_label=task_node.label,
                    data_coordinate=list(skeleton.get_data_id(quantum_key).full_values),
                    inputs=self._make_predicted_datasets(
                        skeleton,
                        task_node.iter_all_inputs(),
                        skeleton.iter_inputs_of(quantum_key),
                    ),
                    outputs=self._make_predicted_datasets(
                        skeleton,
                        task_node.iter_all_outputs(),
                        skeleton.iter_outputs_of(quantum_key),
                    ),
                    datastore_records={
                        datastore_name: records.to_simple()
                        for datastore_name, records in skeleton[quantum_key]
                        .get("datastore_records", {})
                        .items()
                    },
                )
                components.quantum_datasets[quantum_datasets.quantum_id] = quantum_datasets
        self.log.verbose("Building the thin summary graph.")
        components.set_thin_graph()
        components.set_header_counts()
        self.log.info("Graph build complete.")
        return components

    @staticmethod
    def _make_predicted_datasets(
        skeleton: QuantumGraphSkeleton,
        edges: Iterable[Edge],
        dataset_keys: Iterable[DatasetKey | PrerequisiteDatasetKey],
    ) -> dict[str, list[PredictedDatasetModel]]:
        """Make the predicted quantum graph model objects that represent the
        datasets from an iterable of pipeline graph edges.

        Parameters
        ----------
        skeleton : `quantum_graph_skeleton.QuantumGraphSkeleton`
            Temporary data structure used by the builder to represent the
            graph.
        edges : `~collections.abc.Iterable` [ `.pipeline_graph.Edge` ]
            Pipeline graph edges.
        dataset_keys : `~collections.abc.Iterable` [ \
                `.quantum_graph_skeleton.DatasetKey` or\
                `.quantum_graph_skeleton.PrerequisiteDatasetKey` ]
            All nodes in the skeleton that correspond to any of the given
            pipeline graph edges.

        Returns
        -------
        predicted_datasets : `dict` [ `str`, \
                `list` [ `.quantum_graph.PredictedDatasetModel` ] ]
            Mapping of dataset models, keyed by connection name.
        """
        from .quantum_graph import PredictedDatasetModel

        connection_names_by_dataset_type: defaultdict[str, list[str]] = defaultdict(list)
        result: dict[str, list[PredictedDatasetModel]] = {}
        for edge in edges:
            connection_names_by_dataset_type[edge.parent_dataset_type_name].append(edge.connection_name)
            result[edge.connection_name] = []

        for dataset_key in dataset_keys:
            connection_names = connection_names_by_dataset_type.get(dataset_key.parent_dataset_type_name)
            if connection_names is None:
                # Ignore if this isn't one of the connections we're processing
                # (probably an init-input), which would also be predecessor to
                # a quantum node, but should be handled separately.
                continue
            ref = skeleton.get_dataset_ref(dataset_key)
            assert ref is not None, "DatasetRefs should have already been added to skeleton."
            for connection_name in connection_names:
                result[connection_name].append(PredictedDatasetModel.from_dataset_ref(ref))
        for refs in result.values():
            refs.sort(key=operator.attrgetter("data_coordinate"))
        return result

    @staticmethod
    @final
    def _find_removed(
        original: Iterable[DatasetKey | PrerequisiteDatasetKey],
        adjusted: NamedKeyMapping[DatasetType, Sequence[DatasetRef]],
    ) -> set[DatasetKey | PrerequisiteDatasetKey]:
        """Identify skeleton-graph dataset nodes that have been removed by
        `~PipelineTaskConnections.adjustQuantum`.

        Parameters
        ----------
        original : `~collections.abc.Iterable` [ `DatasetKey` or \
                `PrerequisiteDatasetKey` ]
            Identifiers for the dataset nodes that were the original neighbors
            (inputs or outputs) of a quantum.
        adjusted : `~lsst.daf.butler.NamedKeyMapping` [ \
                `~lsst.daf.butler.DatasetType`, \
                `~collections.abc.Sequence` [ `lsst.daf.butler.DatasetType` ] ]
            Adjusted neighbors, in the form used by `lsst.daf.butler.Quantum`.

        Returns
        -------
        removed : `set` [ `DatasetKey` ]
            Datasets in ``original`` that have no counterpart in ``adjusted``.
        """
        result = set(original)
        for dataset_type, kept_refs in adjusted.items():
            parent_dataset_type_name, _ = DatasetType.splitDatasetTypeName(dataset_type.name)
            for kept_ref in kept_refs:
                # We don't know if this was a DatasetKey or a
                # PrerequisiteDatasetKey; just try both.
                result.discard(DatasetKey(parent_dataset_type_name, kept_ref.dataId.required_values))
                result.discard(PrerequisiteDatasetKey(parent_dataset_type_name, kept_ref.id.bytes))
        return result


@dataclasses.dataclass(eq=False, order=False)
class EmptyDimensionsDatasets:
    """Struct that holds the results of empty-dimensions dataset queries for
    `QuantumGraphBuilder`.
    """

    inputs: Mapping[DatasetKey | PrerequisiteDatasetKey, DatasetRef] = dataclasses.field(default_factory=dict)
    """Overall-input datasets found in `QuantumGraphBuilder.input_collections`.

    This may include prerequisite inputs.  It does include init-inputs.
    It does not include intermediates.
    """

    outputs_for_skip: Mapping[DatasetKey, DatasetRef] = dataclasses.field(default_factory=dict)
    """Output datasets found in `QuantumGraphBuilder.skip_existing_in`.

    It is unspecified whether this contains init-outputs; there is
    no concept of skipping at the init stage, so this is not expected to
    matter.
    """

    outputs_in_the_way: Mapping[DatasetKey, DatasetRef] = dataclasses.field(default_factory=dict)
    """Output datasets found in `QuantumGraphBuilder.output_run`.

    This includes regular outputs and init-outputs.
    """


def _quantum_or_quanta(n: int) -> str:
    """Correctly pluralize 'quantum' if needed."""
    return f"{n} quanta" if n != 1 else "1 quantum"