Improve error message for duplicate pipeline run names

docs/book/how-to/steps-pipelines/yaml_configuration.md

@@ -116,6 +116,25 @@ Set a custom name for the pipeline run:
 run_name: "training_run_cifar10_resnet50_lr0.001"
 ```
 
+{% hint style="warning" %}
+**Important:** Pipeline run names must be unique within a project. If you try to run a pipeline with a name that already exists, you'll get an error. To avoid this:
+
+1. **Use dynamic placeholders** to ensure uniqueness:
+   ```yaml
+   # Example 1: Use placeholders for date and time to ensure uniqueness
+   run_name: "training_run_{date}_{time}"
+
+   # Example 2: Combine placeholders with specific details for better context
+   run_name: "training_run_cifar10_resnet50_lr0.001_{date}_{time}"
+   ```
+
+2. **Remove the 'run_name' from your config** to let ZenML auto-generate unique names
+
+3. **Change the run_name** before rerunning the pipeline
+
+Available placeholders: `{date}`, `{time}`, and any parameters defined in your pipeline configuration.
+{% endhint %}
+
 ## Resource and Component Configuration
 
 ### Docker Settings

src/zenml/zen_stores/sql_zen_store.py

@@ -5664,6 +5664,27 @@ def _get_regular_output_artifact_node(
             pipeline_run_id=pipeline_run_id, status=ExecutionStatus(run.status)
         )
 
+    def _get_duplicate_run_name_error_message(
+        self, pipeline_run_name: str
+    ) -> str:
+        """Generate a user-friendly error message for duplicate pipeline run names.
+
+        Args:
+            pipeline_run_name: The name of the pipeline run that already exists.
+
+        Returns:
+            A formatted error message with helpful suggestions.
+        """
+        return (
+            f"Pipeline run name '{pipeline_run_name}' already exists in this project. "
+            f"Each pipeline run must have a unique name.\n\n"
+            f"To fix this, you can:\n"
+            f"1. Use a different run name\n"
+            f'2. Use a dynamic run name with placeholders like: "{pipeline_run_name}_{{date}}_{{time}}"\n'
+            f"3. Remove the run name from your configuration to auto-generate unique names\n\n"
+            f"For more information on run naming, see: https://docs.zenml.io/concepts/steps_and_pipelines/yaml_configuration#run-name"
+        )
+
     def _create_run(
         self, pipeline_run: PipelineRunRequest, session: Session
     ) -> PipelineRunResponse:
@@ -5700,27 +5721,29 @@ def _create_run(
 
         # Add logs entry for the run if exists
         if pipeline_run.logs is not None:
-            self._get_reference_schema_by_id(
-                resource=pipeline_run,
-                reference_schema=StackComponentSchema,
-                reference_id=pipeline_run.logs.artifact_store_id,
-                session=session,
-                reference_type="logs artifact store",
-            )
+            with session.no_autoflush:
+                self._get_reference_schema_by_id(
+                    resource=pipeline_run,
+                    reference_schema=StackComponentSchema,
+                    reference_id=pipeline_run.logs.artifact_store_id,
+                    session=session,
+                    reference_type="logs artifact store",
+                )
 
-            log_entry = LogsSchema(
-                uri=pipeline_run.logs.uri,
-                pipeline_run_id=new_run.id,
-                artifact_store_id=pipeline_run.logs.artifact_store_id,
-            )
-            session.add(log_entry)
+                log_entry = LogsSchema(
+                    uri=pipeline_run.logs.uri,
+                    pipeline_run_id=new_run.id,
+                    artifact_store_id=pipeline_run.logs.artifact_store_id,
+                )
+                session.add(log_entry)
 
         try:
             session.commit()
         except IntegrityError:
             # We have to rollback the failed session first in order to
             # continue using it
             session.rollback()
+
             # This can fail if the name is taken by a different run
             self._verify_name_uniqueness(
                 resource=pipeline_run,
@@ -6020,7 +6043,14 @@ def get_or_create_run(
                 return self._create_run(pipeline_run, session=session), True
             except EntityExistsError as create_error:
                 if not pipeline_run.orchestrator_run_id:
-                    raise
+                    # No orchestrator_run_id means this is likely a name conflict.
+                    # Provide a user-friendly error message for duplicate run names.
+                    improved_message = (
+                        self._get_duplicate_run_name_error_message(
+                            pipeline_run.name
+                        )
+                    )
+                    raise EntityExistsError(improved_message) from create_error
                 # Creating the run failed because
                 # - a run with the same deployment_id and orchestrator_run_id
                 #   exists. We now fetch and return that run.
@@ -6039,10 +6069,14 @@ def get_or_create_run(
                     )
                 except KeyError:
                     # We should only get here if the run creation failed because
-                    # of a name conflict. We raise the error that happened
-                    # during creation in any case to forward the error message
-                    # to the user.
-                    raise create_error
+                    # of a name conflict. Provide a user-friendly error message
+                    # for duplicate run names.
+                    improved_message = (
+                        self._get_duplicate_run_name_error_message(
+                            pipeline_run.name
+                        )
+                    )
+                    raise EntityExistsError(improved_message) from create_error
 
     def list_runs(
         self,

tests/integration/functional/pipelines/test_pipeline_run.py

@@ -1,8 +1,11 @@
 import os
 from unittest.mock import patch
 
+import pytest
+
 from zenml import pipeline, step
 from zenml.constants import ENV_ZENML_PREVENT_CLIENT_SIDE_CACHING
+from zenml.exceptions import EntityExistsError
 
 
 @step(enable_cache=False)
@@ -29,6 +32,11 @@ def noop() -> None:
     pass
 
 
+@step(enable_cache=False)
+def simple_step_for_duplicate_test() -> int:
+    return 42
+
+
 def test_pipeline_run_computes_clientside_cache(clean_client, mocker):
     """Tests that running a pipeline computes the cached steps client-side and
     only forwards the non-cached steps to the orchestrator.
@@ -104,3 +112,29 @@ def full_cached_pipeline():
         full_cached_pipeline()
 
     mock_submit_pipeline.assert_called()
+
+
+def test_duplicate_pipeline_run_name_raises_improved_error(clean_client):
+    """Test that running a pipeline twice with the same name raises an improved error message."""
+
+    @pipeline
+    def test_pipeline():
+        simple_step_for_duplicate_test()
+
+    # First run should succeed
+    run_name = "duplicate_name_test_run"
+    first_run = test_pipeline.with_options(run_name=run_name)()
+    assert first_run.name == run_name
+
+    # Second run with same name should raise EntityExistsError with clear message
+    with pytest.raises(EntityExistsError) as exc_info:
+        test_pipeline.with_options(run_name=run_name)()
+
+    error_message = str(exc_info.value)
+
+    # Verify it contains a clear duplicate name error message
+    assert (
+        "already exists" in error_message.lower()
+        or "existing pipeline run with the same name" in error_message.lower()
+        or f"Pipeline run name '{run_name}' already exists" in error_message
+    )