Coverage for python / lsst / pipe / base / single_quantum_executor.py: 11%
219 statements
« prev ^ index » next coverage.py v7.12.0, created at 2025-12-12 01:14 -0800
1# This file is part of pipe_base.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (http://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# This software is dual licensed under the GNU General Public License and also
10# under a 3-clause BSD license. Recipients may choose which of these licenses
11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt,
12# respectively. If you choose the GPL option then the following text applies
13# (but note that there is still no warranty even if you opt for BSD instead):
14#
15# This program is free software: you can redistribute it and/or modify
16# it under the terms of the GNU General Public License as published by
17# the Free Software Foundation, either version 3 of the License, or
18# (at your option) any later version.
19#
20# This program is distributed in the hope that it will be useful,
21# but WITHOUT ANY WARRANTY; without even the implied warranty of
22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
23# GNU General Public License for more details.
24#
25# You should have received a copy of the GNU General Public License
26# along with this program. If not, see <http://www.gnu.org/licenses/>.
28__all__ = ["SingleQuantumExecutor"]
30import logging
31import time
32import uuid
33from collections import defaultdict
34from collections.abc import Callable, Mapping
35from itertools import chain
36from typing import Any
38from lsst.daf.butler import (
39 Butler,
40 ButlerMetrics,
41 DatasetRef,
42 DatasetType,
43 LimitedButler,
44 NamedKeyDict,
45 Quantum,
46)
47from lsst.utils.introspection import get_full_type_name
48from lsst.utils.timer import logInfo
50from ._quantumContext import ExecutionResources, QuantumContext
51from ._status import (
52 AnnotatedPartialOutputsError,
53 ExceptionInfo,
54 InvalidQuantumError,
55 NoWorkFound,
56 QuantumSuccessCaveats,
57)
58from .connections import AdjustQuantumHelper
59from .log_capture import LogCapture, _ExecutionLogRecordsExtra
60from .pipeline_graph import TaskNode
61from .pipelineTask import PipelineTask
62from .quantum_graph_executor import QuantumExecutor
63from .quantum_reports import QuantumReport
64from .task import _TASK_FULL_METADATA_TYPE, _TASK_METADATA_TYPE
65from .taskFactory import TaskFactory
67_LOG = logging.getLogger(__name__)
70class SingleQuantumExecutor(QuantumExecutor):
71 """Executor class which runs one Quantum at a time.
73 Parameters
74 ----------
75 butler : `~lsst.daf.butler.Butler` or `None`, optional
76 Data butler, `None` means that a limited butler should be used instead.
77 task_factory : `.TaskFactory`, optional
78 Instance of a task factory. Defaults to a new instance of
79 `lsst.pipe.base.TaskFactory`.
80 skip_existing_in : `str` or `~collections.abc.Iterable` [ `str` ]
81 Expressions representing the collections to search for existing output
82 datasets. See :ref:`daf_butler_ordered_collection_searches` for allowed
83 types. This class only checks for the presence of butler output run in
84 the list of collections. If the output run is present in the list then
85 the quanta whose complete outputs exist in the output run will be
86 skipped. `None` or empty string/sequence disables skipping.
87 clobber_outputs : `bool`, optional
88 If `True`, then outputs from a quantum that exist in output run
89 collection will be removed prior to executing a quantum. If
90 ``skip_existing_in`` contains output run, then only partial outputs
91 from a quantum will be removed. Only used when ``butler`` is not
92 `None`.
93 enable_lsst_debug : `bool`, optional
94 Enable debugging with ``lsstDebug`` facility for a task.
95 limited_butler_factory : `Callable`, optional
96 A method that creates a `~lsst.daf.butler.LimitedButler` instance for a
97 given Quantum. This parameter must be defined if ``butler`` is `None`.
98 If ``butler`` is not `None` then this parameter is ignored.
99 resources : `.ExecutionResources`, optional
100 The resources available to this quantum when executing.
101 skip_existing : `bool`, optional
102 If `True`, skip quanta whose metadata datasets are already stored.
103 Unlike ``skip_existing_in``, this works with limited butlers as well as
104 full butlers. Always set to `True` if ``skip_existing_in`` matches
105 ``butler.run``.
106 assume_no_existing_outputs : `bool`, optional
107 If `True`, assume preexisting outputs are impossible (e.g. because this
108 is known by higher-level code to be a new ``RUN`` collection), and do
109 not look for them. This causes the ``skip_existing`` and
110 ``clobber_outputs`` options to be ignored, but unlike just setting both
111 of those to `False`, it also avoids all dataset existence checks.
112 raise_on_partial_outputs : `bool`, optional
113 If `True` raise exceptions chained by `.AnnotatedPartialOutputsError`
114 immediately, instead of considering the partial result a success and
115 continuing to run downstream tasks.
116 job_metadata : `~collections.abc.Mapping`
117 Mapping with extra metadata to embed within the quantum metadata under
118 the "job" key. This is intended to correspond to information common
119 to all quanta being executed in a single process, such as the time
120 taken to load the quantum graph in a BPS job.
121 """
123 def __init__(
124 self,
125 *,
126 butler: Butler | None = None,
127 task_factory: TaskFactory | None = None,
128 skip_existing_in: Any = None,
129 clobber_outputs: bool = False,
130 enable_lsst_debug: bool = False,
131 limited_butler_factory: Callable[[Quantum], LimitedButler] | None = None,
132 resources: ExecutionResources | None = None,
133 skip_existing: bool = False,
134 assume_no_existing_outputs: bool = False,
135 raise_on_partial_outputs: bool = True,
136 job_metadata: Mapping[str, int | str | float] | None = None,
137 ):
138 self._butler = butler
139 self._task_factory = task_factory if task_factory is not None else TaskFactory()
140 self._clobber_outputs = clobber_outputs
141 self._enable_lsst_debug = enable_lsst_debug
142 self._limited_butler_factory = limited_butler_factory
143 self._resources = resources
144 self._assume_no_existing_outputs = assume_no_existing_outputs
145 self._raise_on_partial_outputs = raise_on_partial_outputs
146 self._job_metadata = job_metadata
148 if self._butler is None:
149 assert limited_butler_factory is not None, "limited_butler_factory is needed when butler is None"
151 # Find whether output run is in skip_existing_in.
152 self._skip_existing = skip_existing
153 if self._butler is not None and skip_existing_in and not self._skip_existing:
154 self._skip_existing = self._butler.run in self._butler.collections.query(
155 skip_existing_in, flatten_chains=True
156 )
157 self._previous_process_quanta: list[uuid.UUID] = []
159 def execute(
160 self, task_node: TaskNode, /, quantum: Quantum, quantum_id: uuid.UUID | None = None
161 ) -> tuple[Quantum, QuantumReport | None]:
162 # Docstring inherited from QuantumExecutor.execute
163 assert quantum.dataId is not None, "Quantum DataId cannot be None"
165 if self._butler is not None:
166 self._butler.registry.refresh()
168 result = self._execute(task_node, quantum, quantum_id=quantum_id)
169 report = QuantumReport(quantumId=quantum_id, dataId=quantum.dataId, taskLabel=task_node.label)
170 return result, report
172 def _execute(
173 self, task_node: TaskNode, /, quantum: Quantum, quantum_id: uuid.UUID | None = None
174 ) -> Quantum:
175 """Execute the quantum.
177 Internal implementation of `execute()`.
178 """
179 # Make a limited butler instance if needed.
180 limited_butler: LimitedButler
181 used_butler_factory = False
182 if self._butler is not None:
183 limited_butler = self._butler
184 else:
185 # We check this in constructor, but mypy needs this check here.
186 assert self._limited_butler_factory is not None
187 limited_butler = self._limited_butler_factory(quantum)
188 used_butler_factory = True
190 try:
191 return self._execute_with_limited_butler(
192 task_node, limited_butler, quantum=quantum, quantum_id=quantum_id
193 )
194 finally:
195 if used_butler_factory:
196 limited_butler.close()
198 def _execute_with_limited_butler(
199 self,
200 task_node: TaskNode,
201 limited_butler: LimitedButler,
202 /,
203 quantum: Quantum,
204 quantum_id: uuid.UUID | None = None,
205 ) -> Quantum:
206 startTime = time.time()
208 if self._butler is not None:
209 log_capture = LogCapture.from_full(self._butler)
210 else:
211 log_capture = LogCapture.from_limited(limited_butler)
212 with log_capture.capture_logging(task_node, quantum) as captureLog:
213 # Save detailed resource usage before task start to metadata.
214 quantumMetadata = _TASK_METADATA_TYPE()
215 logInfo(None, "prep", metadata=quantumMetadata) # type: ignore[arg-type]
217 _LOG.info(
218 "Preparing execution of quantum for label=%s dataId=%s.", task_node.label, quantum.dataId
219 )
221 # check whether to skip or delete old outputs, if it returns True
222 # or raises an exception do not try to store logs, as they may be
223 # already in butler.
224 captureLog.store = False
225 if self._check_existing_outputs(quantum, task_node, limited_butler, captureLog.extra):
226 _LOG.info(
227 "Skipping already-successful quantum for label=%s dataId=%s.",
228 task_node.label,
229 quantum.dataId,
230 )
231 return quantum
232 captureLog.store = True
234 captureLog.extra.previous_process_quanta.extend(self._previous_process_quanta)
235 if quantum_id is not None:
236 self._previous_process_quanta.append(quantum_id)
237 try:
238 quantum = self._updated_quantum_inputs(quantum, task_node, limited_butler)
239 except NoWorkFound as exc:
240 _LOG.info(
241 "Nothing to do for task '%s' on quantum %s; saving metadata and skipping: %s",
242 task_node.label,
243 quantum.dataId,
244 str(exc),
245 )
246 quantumMetadata["caveats"] = QuantumSuccessCaveats.from_adjust_quantum_no_work().value
247 # Make empty metadata that looks something like what a
248 # do-nothing task would write (but we don't bother with empty
249 # nested PropertySets for subtasks). This is slightly
250 # duplicative with logic in pipe_base that we can't easily call
251 # from here; we'll fix this on DM-29761.
252 logInfo(None, "end", metadata=quantumMetadata) # type: ignore[arg-type]
253 fullMetadata = _TASK_FULL_METADATA_TYPE()
254 fullMetadata[task_node.label] = _TASK_METADATA_TYPE()
255 fullMetadata["quantum"] = quantumMetadata
256 if self._job_metadata is not None:
257 fullMetadata["job"] = self._job_metadata
258 self._write_metadata(quantum, fullMetadata, task_node, limited_butler)
259 return quantum
261 # enable lsstDebug debugging
262 if self._enable_lsst_debug:
263 try:
264 _LOG.debug("Will try to import debug.py")
265 import debug # type: ignore # noqa:F401
266 except ImportError:
267 _LOG.warning("No 'debug' module found.")
269 # Ensure that we are executing a frozen config
270 task_node.config.freeze()
271 logInfo(None, "init", metadata=quantumMetadata) # type: ignore[arg-type]
272 init_input_refs = list(quantum.initInputs.values())
274 _LOG.info(
275 "Constructing task and executing quantum for label=%s dataId=%s.",
276 task_node.label,
277 quantum.dataId,
278 )
279 task = self._task_factory.makeTask(task_node, limited_butler, init_input_refs)
280 logInfo(None, "start", metadata=quantumMetadata) # type: ignore[arg-type]
281 try:
282 caveats, outputsPut, butler_metrics = self._run_quantum(
283 task, quantum, task_node, limited_butler, quantum_id=quantum_id
284 )
285 except Exception as e:
286 _LOG.error(
287 "Execution of task '%s' on quantum %s failed. Exception %s: %s",
288 task_node.label,
289 quantum.dataId,
290 e.__class__.__name__,
291 str(e),
292 )
293 captureLog.extra.exception = ExceptionInfo(
294 type_name=get_full_type_name(e),
295 message=str(e),
296 metadata={},
297 )
298 raise
299 else:
300 quantumMetadata["butler_metrics"] = butler_metrics.model_dump()
301 quantumMetadata["caveats"] = caveats.value
302 # Stringify the UUID for easier compatibility with
303 # PropertyList.
304 quantumMetadata["outputs"] = [str(output) for output in outputsPut]
305 finally:
306 logInfo(None, "end", metadata=quantumMetadata) # type: ignore[arg-type]
307 fullMetadata = task.getFullMetadata()
308 fullMetadata["quantum"] = quantumMetadata
309 if self._job_metadata is not None:
310 fullMetadata["job"] = self._job_metadata
311 captureLog.extra.metadata = fullMetadata
312 self._write_metadata(quantum, fullMetadata, task_node, limited_butler)
313 stopTime = time.time()
314 _LOG.info(
315 "Execution of task '%s' on quantum %s took %.3f seconds",
316 task_node.label,
317 quantum.dataId,
318 stopTime - startTime,
319 )
320 return quantum
322 def _check_existing_outputs(
323 self,
324 quantum: Quantum,
325 task_node: TaskNode,
326 /,
327 limited_butler: LimitedButler,
328 log_extra: _ExecutionLogRecordsExtra,
329 ) -> bool:
330 """Decide whether this quantum needs to be executed.
332 If only partial outputs exist then they are removed if
333 ``clobberOutputs`` is True, otherwise an exception is raised.
335 The ``LimitedButler`` is used for everything, and should be set to
336 ``self.butler`` if no separate ``LimitedButler`` is available.
338 Parameters
339 ----------
340 quantum : `~lsst.daf.butler.Quantum`
341 Quantum to check for existing outputs.
342 task_node : `~.pipeline_graph.TaskNode`
343 Task definition structure.
344 limited_butler : `~lsst.daf.butler.LimitedButler`
345 Butler to use for querying and clobbering.
346 log_extra : `.log_capture.TaskLogRecordsExtra`
347 Extra information to attach to log records.
349 Returns
350 -------
351 exist : `bool`
352 `True` if ``self.skipExisting`` is defined, and a previous
353 execution of this quanta appears to have completed successfully
354 (either because metadata was written or all datasets were written).
355 `False` otherwise.
357 Raises
358 ------
359 RuntimeError
360 Raised if some outputs exist and some not.
361 """
362 if self._assume_no_existing_outputs:
363 return False
365 if self._skip_existing:
366 _LOG.debug(
367 "Checking existence of metadata from previous execution of label=%s dataId=%s.",
368 task_node.label,
369 quantum.dataId,
370 )
371 # Metadata output exists; this is sufficient to assume the previous
372 # run was successful and should be skipped.
373 [metadata_ref] = quantum.outputs[task_node.metadata_output.dataset_type_name]
374 if metadata_ref is not None:
375 if limited_butler.stored(metadata_ref):
376 return True
378 # Find and prune (partial) outputs if `self.clobberOutputs` is set.
379 _LOG.debug(
380 "Looking for existing outputs in the way for label=%s dataId=%s.", task_node.label, quantum.dataId
381 )
382 ref_dict = limited_butler.stored_many(chain.from_iterable(quantum.outputs.values()))
383 if task_node.log_output is not None:
384 (log_ref,) = quantum.outputs[task_node.log_output.dataset_type_name]
385 if ref_dict[log_ref]:
386 _LOG.debug(
387 "Attaching logs from previous attempt on label=%s dataId=%s.",
388 task_node.label,
389 quantum.dataId,
390 )
391 log_extra.attach_previous_attempt(limited_butler.get(log_ref))
392 existingRefs = [ref for ref, exists in ref_dict.items() if exists]
393 missingRefs = [ref for ref, exists in ref_dict.items() if not exists]
394 if existingRefs:
395 if not missingRefs:
396 # Full outputs exist.
397 if self._skip_existing:
398 return True
399 elif self._clobber_outputs:
400 _LOG.info("Removing complete outputs for quantum %s: %s", quantum, existingRefs)
401 limited_butler.pruneDatasets(existingRefs, disassociate=True, unstore=True, purge=True)
402 else:
403 raise RuntimeError(
404 f"Complete outputs exists for a quantum {quantum} "
405 "and neither clobberOutputs nor skipExisting is set: "
406 f"existingRefs={existingRefs}"
407 )
408 else:
409 # Partial outputs from a failed quantum.
410 _LOG.debug(
411 "Partial outputs exist for quantum %s existingRefs=%s missingRefs=%s",
412 quantum,
413 existingRefs,
414 missingRefs,
415 )
416 if self._clobber_outputs:
417 # only prune
418 _LOG.info("Removing partial outputs for task %s: %s", task_node.label, existingRefs)
419 limited_butler.pruneDatasets(existingRefs, disassociate=True, unstore=True, purge=True)
420 return False
421 else:
422 raise RuntimeError(
423 "Registry inconsistency while checking for existing quantum outputs:"
424 f" quantum={quantum} existingRefs={existingRefs}"
425 f" missingRefs={missingRefs}"
426 )
428 # By default always execute.
429 return False
431 def _updated_quantum_inputs(
432 self, quantum: Quantum, task_node: TaskNode, /, limited_butler: LimitedButler
433 ) -> Quantum:
434 """Update quantum with extra information, returns a new updated
435 Quantum.
437 Some methods may require input DatasetRefs to have non-None
438 ``dataset_id``, but in case of intermediate dataset it may not be
439 filled during QuantumGraph construction. This method will retrieve
440 missing info from registry.
442 Parameters
443 ----------
444 quantum : `~lsst.daf.butler.Quantum`
445 Single Quantum instance.
446 task_node : `~.pipeline_graph.TaskNode`
447 Task definition structure.
448 limited_butler : `~lsst.daf.butler.LimitedButler`
449 Butler to use for querying.
451 Returns
452 -------
453 update : `~lsst.daf.butler.Quantum`
454 Updated Quantum instance.
455 """
456 anyChanges = False
457 updatedInputs: defaultdict[DatasetType, list] = defaultdict(list)
458 for key, refsForDatasetType in quantum.inputs.items():
459 _LOG.debug(
460 "Checking existence of input '%s' for label=%s dataId=%s.",
461 key.name,
462 task_node.label,
463 quantum.dataId,
464 )
465 toCheck = []
466 newRefsForDatasetType = updatedInputs[key]
467 for ref in refsForDatasetType:
468 if self._should_assume_exists(quantum, ref):
469 newRefsForDatasetType.append(ref)
470 else:
471 toCheck.append(ref)
472 if not toCheck:
473 _LOG.debug(
474 "Assuming overall input '%s' is present without checks for label=%s dataId=%s.",
475 key.name,
476 task_node.label,
477 quantum.dataId,
478 )
479 continue
480 stored = limited_butler.stored_many(toCheck)
481 for ref in toCheck:
482 if stored[ref]:
483 newRefsForDatasetType.append(ref)
484 else:
485 # This should only happen if a predicted intermediate was
486 # not actually produced upstream, but
487 # datastore misconfigurations can unfortunately also land
488 # us here.
489 _LOG.info("No dataset artifact found for %s", ref)
490 continue
491 if len(newRefsForDatasetType) != len(refsForDatasetType):
492 anyChanges = True
493 # If we removed any input datasets, let the task check if it has enough
494 # to proceed and/or prune related datasets that it also doesn't
495 # need/produce anymore. It will raise NoWorkFound if it can't run,
496 # which we'll let propagate up. This is exactly what we run during QG
497 # generation, because a task shouldn't care whether an input is missing
498 # because some previous task didn't produce it, or because it just
499 # wasn't there during QG generation.
500 namedUpdatedInputs = NamedKeyDict[DatasetType, list[DatasetRef]](updatedInputs.items())
501 helper = AdjustQuantumHelper(namedUpdatedInputs, quantum.outputs)
502 if anyChanges:
503 _LOG.debug("Running adjustQuantum for label=%s dataId=%s.", task_node.label, quantum.dataId)
504 assert quantum.dataId is not None, "Quantum DataId cannot be None"
505 helper.adjust_in_place(task_node.get_connections(), label=task_node.label, data_id=quantum.dataId)
506 return Quantum(
507 taskName=quantum.taskName,
508 taskClass=quantum.taskClass,
509 dataId=quantum.dataId,
510 initInputs=quantum.initInputs,
511 inputs=helper.inputs,
512 outputs=helper.outputs,
513 )
515 def _run_quantum(
516 self,
517 task: PipelineTask,
518 quantum: Quantum,
519 task_node: TaskNode,
520 /,
521 limited_butler: LimitedButler,
522 quantum_id: uuid.UUID | None = None,
523 ) -> tuple[QuantumSuccessCaveats, list[uuid.UUID], ButlerMetrics]:
524 """Execute task on a single quantum.
526 Parameters
527 ----------
528 task : `PipelineTask`
529 Task object.
530 quantum : `~lsst.daf.butler.Quantum`
531 Single Quantum instance.
532 task_node : `~.pipeline_graph.TaskNode`
533 Task definition structure.
534 limited_butler : `~lsst.daf.butler.LimitedButler`
535 Butler to use for dataset I/O.
536 quantum_id : `uuid.UUID` or `None`, optional
537 ID of the quantum being executed.
539 Returns
540 -------
541 flags : `QuantumSuccessCaveats`
542 Flags that describe qualified successes.
543 ids_put : list[ `uuid.UUID` ]
544 Record of all the dataset IDs that were written by this quantum
545 being executed.
546 metrics : `lsst.daf.butler.ButlerMetrics`
547 Butler metrics recorded for this quantum.
548 """
549 flags = QuantumSuccessCaveats.NO_CAVEATS
551 # Create a butler that operates in the context of a quantum
552 butlerQC = QuantumContext(limited_butler, quantum, resources=self._resources, quantum_id=quantum_id)
554 # Get the input and output references for the task
555 inputRefs, outputRefs = task_node.get_connections().buildDatasetRefs(quantum)
557 # Call task runQuantum() method.
558 try:
559 with limited_butler.record_metrics() as butler_metrics:
560 task.runQuantum(butlerQC, inputRefs, outputRefs)
561 except NoWorkFound as err:
562 # Not an error, just an early exit.
563 _LOG.info(
564 "Task '%s' on quantum %s exited early with no work found: %s.",
565 task_node.label,
566 quantum.dataId,
567 str(err),
568 )
569 flags |= err.FLAGS
570 except AnnotatedPartialOutputsError as caught:
571 error: BaseException
572 if caught.__cause__ is None:
573 _LOG.error(
574 "Incorrect use of AnnotatedPartialOutputsError: no chained exception found.",
575 task_node.label,
576 quantum.dataId,
577 )
578 error = caught
579 else:
580 error = caught.__cause__
581 if self._raise_on_partial_outputs:
582 # Note: this is a real edge case that required some
583 # experimentation: without 'from None' below, this raise would
584 # produce a "while one exception was being handled, another was
585 # raised" traceback involving AnnotatedPartialOutputsError.
586 # With the 'from None', we get just the error chained to it, as
587 # desired.
588 raise error from None
589 else:
590 _LOG.error(
591 "Task '%s' on quantum %s exited with partial outputs; "
592 "considering this a qualified success and proceeding.",
593 task_node.label,
594 quantum.dataId,
595 )
596 _LOG.error(error, exc_info=error)
597 flags |= caught.FLAGS
598 if not butlerQC.outputsPut:
599 flags |= QuantumSuccessCaveats.ALL_OUTPUTS_MISSING
600 if not butlerQC.outputsPut == butlerQC.allOutputs:
601 flags |= QuantumSuccessCaveats.ANY_OUTPUTS_MISSING
602 ids_put = [output[2] for output in butlerQC.outputsPut]
603 return flags, ids_put, butler_metrics
605 def _write_metadata(
606 self, quantum: Quantum, metadata: Any, task_node: TaskNode, /, limited_butler: LimitedButler
607 ) -> None:
608 # DatasetRef has to be in the Quantum outputs, can lookup by name
609 try:
610 [ref] = quantum.outputs[task_node.metadata_output.dataset_type_name]
611 except LookupError as exc:
612 raise InvalidQuantumError(
613 "Quantum outputs is missing metadata dataset type "
614 f"{task_node.metadata_output.dataset_type_name};"
615 " this could happen due to inconsistent options between QuantumGraph generation"
616 " and execution"
617 ) from exc
618 limited_butler.put(metadata, ref)
620 def _should_assume_exists(self, quantum: Quantum, ref: DatasetRef) -> bool | None:
621 """Report whether the given dataset can be assumed to exist because
622 some previous check reported that it did.
624 If this is `True` for a dataset does not in fact exist anymore, that's
625 an unexpected problem that we want to raise as an exception, and
626 definitely not a case where some predicted output just wasn't produced.
627 We can't always tell the difference, but in this case we can.
629 Parameters
630 ----------
631 quantum : `Quantum`
632 Quantum being processed.
633 ref : `lsst.daf.butler.DatasetRef`
634 Reference to the input dataset.
636 Returns
637 -------
638 exists : `bool` or `None`
639 `True` if this dataset is definitely an overall input, `False` if
640 some other quantum in the graph is expected to produce it, and
641 `None` if the answer could not be determined.
642 """
643 if quantum.datastore_records:
644 for datastore_record_data in quantum.datastore_records.values():
645 if ref.id in datastore_record_data.records:
646 return True
647 return False
648 return None