Repository: kedro-org/kedro
File: kedro/runner/runner.py (Python)

Code before:
        # Check which datasets used in the pipeline are in the catalog or match
        # a pattern in the catalog, not including extra dataset patterns
        registered_ds = [ds for ds in pipeline.datasets() if ds in catalog]

        # Check if there are any input datasets that aren't in the catalog and
        # don't match a pattern in the catalog.
        unsatisfied = pipeline.inputs() - set(registered_ds)

        if unsatisfied:
            raise ValueError(
                f"Pipeline input(s) {unsatisfied} not found in the {catalog.__class__.__name__}"
            )

        # Register the default dataset pattern with the catalog
        catalog = catalog.shallow_copy(
            extra_dataset_patterns=self._extra_dataset_patterns
        )

        hook_or_null_manager = hook_manager or _NullPluginManager()

        # Check which datasets used in the pipeline are in the catalog or match
        # a pattern in the catalog, including added extra_dataset_patterns
        registered_ds = [ds for ds in pipeline.datasets() if ds in catalog]

        if self._is_async:
            self._logger.info(
                "Asynchronous mode is enabled for loading and saving data"
            )

        for ds in pipeline.datasets():
            _ = catalog._get_dataset(ds)

        self._run(pipeline, catalog, hook_or_null_manager, session_id)  # type: ignore[arg-type]

        self._logger.info("Pipeline execution completed successfully.")

        # Identify MemoryDataset in the catalog
        memory_datasets = {
            ds_name
            for ds_name, ds in catalog._datasets.items()
            if isinstance(ds, MemoryDataset) or isinstance(ds, SharedMemoryDataset)
        }

        # Check if there's any output datasets that aren't in the catalog and don't match a pattern
        # in the catalog and include MemoryDataset.
        free_outputs = pipeline.outputs() - (set(registered_ds) - memory_datasets)

        run_output = {ds_name: catalog.load(ds_name) for ds_name in free_outputs}

        return run_output


Review comment (refactor):
Can we add a comment about this? I also notice this line is duplicated:
`registered_ds = [ds for ds in pipeline.datasets() if ds in catalog]`

Maybe we can refactor this into one loop so we only loop `pipeline.datasets()` once? 

Code after:

        # Check if there are any input datasets that aren't in the catalog and
        # don't match a pattern in the catalog.
        unsatisfied = pipeline.inputs() - set(warmed_up_ds)

        if unsatisfied:
            raise ValueError(
                f"Pipeline input(s) {unsatisfied} not found in the {catalog.__class__.__name__}"
            )

        # Register the default dataset pattern with the catalog
        catalog = catalog.shallow_copy(
            extra_dataset_patterns=self._extra_dataset_patterns
        )

        hook_or_null_manager = hook_manager or _NullPluginManager()

        # Check which datasets used in the pipeline are in the catalog or match
        # a pattern in the catalog, including added extra_dataset_patterns
        registered_ds = [ds for ds in pipeline.datasets() if ds in catalog]

        if self._is_async:
            self._logger.info(
                "Asynchronous mode is enabled for loading and saving data"
            )

        self._run(pipeline, catalog, hook_or_null_manager, session_id)  # type: ignore[arg-type]

        self._logger.info("Pipeline execution completed successfully.")

        # Identify MemoryDataset in the catalog
        memory_datasets = {
            ds_name
            for ds_name, ds in catalog._datasets.items()
            if isinstance(ds, MemoryDataset) or isinstance(ds, SharedMemoryDataset)
        }

        # Check if there's any output datasets that aren't in the catalog and don't match a pattern
        # in the catalog and include MemoryDataset.
        free_outputs = pipeline.outputs() - (set(registered_ds) - memory_datasets)

        run_output = {ds_name: catalog.load(ds_name) for ds_name in free_outputs}

        return run_output

    def run_only_missing(
        self, pipeline: Pipeline, catalog: CatalogProtocol, hook_manager: PluginManager
    ) -> dict[str, Any]:
        """Run only the missing outputs from the ``Pipeline`` using the
        datasets provided by ``catalog``, and save results back to the
        same objects.