Move warm-up from session to runner

kedro/framework/session/session.py

@@ -24,7 +24,7 @@
     validate_settings,
 )
 from kedro.io.core import generate_timestamp
-from kedro.runner import AbstractRunner, SequentialRunner, ThreadRunner
+from kedro.runner import AbstractRunner, SequentialRunner
 from kedro.utils import _find_kedro_project
 
 if TYPE_CHECKING:
@@ -395,11 +395,6 @@ def run(  # noqa: PLR0913
         hook_manager.hook.before_pipeline_run(
             run_params=record_data, pipeline=filtered_pipeline, catalog=catalog
         )
-
-        if isinstance(runner, ThreadRunner):
-            for ds in filtered_pipeline.datasets():
-                if catalog.config_resolver.match_pattern(ds):
-                    _ = catalog._get_dataset(ds)
         try:
             run_result = runner.run(
                 filtered_pipeline, catalog, hook_manager, session_id

kedro/runner/runner.py

@@ -112,6 +112,10 @@ def run(
             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.")

tests/runner/test_thread_runner.py

@@ -6,8 +6,15 @@
 import pytest
 
 from kedro.framework.hooks import _create_hook_manager
-from kedro.io import AbstractDataset, DataCatalog, DatasetError, MemoryDataset
+from kedro.io import (
+    AbstractDataset,
+    DataCatalog,
+    DatasetError,
+    KedroDataCatalog,
+    MemoryDataset,
+)
 from kedro.pipeline import node
+from kedro.pipeline.modular_pipeline import pipeline
 from kedro.pipeline.modular_pipeline import pipeline as modular_pipeline
 from kedro.runner import ThreadRunner
 from tests.runner.conftest import exception_fn, identity, return_none, sink, source
@@ -39,6 +46,31 @@ def test_does_not_log_not_using_async(self, fan_out_fan_in, catalog, caplog):
         ThreadRunner().run(fan_out_fan_in, catalog)
         assert "Using synchronous mode for loading and saving data." not in caplog.text
 
+    @pytest.mark.parametrize("catalog_type", [DataCatalog, KedroDataCatalog])
+    def test_thread_run_with_patterns(self, catalog_type):
+        """Test warm-up is done and patterns are resolved before running pipeline.
+
+        Without the warm-up "Dataset 'dummy_1' has already been registered" error
+        would be raised for this test. We check that the dataset was registered at the
+        warm-up, and we successfully passed to loading it.
+        """
+        catalog_conf = {"{catch_all}": {"type": "MemoryDataset"}}
+
+        catalog = catalog_type.from_config(catalog_conf)
+
+        test_pipeline = pipeline(
+            [
+                node(identity, inputs="dummy_1", outputs="output_1", name="node_1"),
+                node(identity, inputs="dummy_2", outputs="output_2", name="node_2"),
+                node(identity, inputs="dummy_1", outputs="output_3", name="node_3"),
+            ]
+        )
+
+        with pytest.raises(
+            Exception, match="Data for MemoryDataset has not been saved yet"
+        ):
+            ThreadRunner().run(test_pipeline, catalog)
+
 
 class TestMaxWorkers:
     @pytest.mark.parametrize(