Avoid retrying successful run operation (#12227)

.changes/unreleased/Fixes-20251128-122838.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: ':bug: :snowman: Avoid retrying successful run-operation commands'
+time: 2025-11-28T12:28:38.546261-05:00
+custom:
+    Author: michelleark
+    Issue: "11850"

core/dbt/task/retry.py

@@ -126,6 +126,7 @@ class RetryTask(ConfiguredTask):
             result.unique_id
             for result in self.previous_results.results
             if result.status in RETRYABLE_STATUSES
+            # Avoid retrying operation nodes unless we are retrying the run-operation command
             and not (
                 self.previous_command_name != "run-operation"
                 and result.unique_id.startswith("operation.")
@@ -150,6 +151,11 @@ class RetryTask(ConfiguredTask):
             )
         }
 
+        # Tasks without get_graph_queue (e.g. run-operation) and no failed nodes to retry.
+        if not unique_ids and not hasattr(self.task_class, "get_graph_queue"):
+            # Return early with the previous results as the past invocation was successful
+            return self.previous_results
+
         class TaskWrapper(self.task_class):
             def get_graph_queue(self):
                 new_graph = self.graph.get_subset_graph(unique_ids)

tests/functional/retry/fixtures.py

@@ -41,6 +41,8 @@ models:
 
 macros__alter_timezone_sql = """
 {% macro alter_timezone(timezone='America/Los_Angeles') %}
+{{ print('running a macro!') }}
+
 {% set sql %}
     SET TimeZone='{{ timezone }}';
 {% endset %}
@@ -50,6 +52,13 @@ macros__alter_timezone_sql = """
 {% endmacro %}
 """
 
+macros__success_macro_sql = """
+{% macro success_macro() %}
+{{ print('running a macro!') }}
+select 1
+{% endmacro %}
+"""
+
 simple_model = """
 select null as id
 """

tests/functional/retry/test_retry.py

@@ -5,9 +5,16 @@ import pytest
 
 from dbt.contracts.results import RunStatus, TestStatus
 from dbt.exceptions import DbtRuntimeError, TargetNotFoundError
-from dbt.tests.util import rm_file, run_dbt, update_config_file, write_file
+from dbt.tests.util import (
+    rm_file,
+    run_dbt,
+    run_dbt_and_capture,
+    update_config_file,
+    write_file,
+)
 from tests.functional.retry.fixtures import (
     macros__alter_timezone_sql,
+    macros__success_macro_sql,
     models__sample_model,
     models__second_model,
     models__union_model,
@@ -64,7 +71,10 @@ class BaseTestRetry:
 
     @pytest.fixture(scope="class")
     def macros(self):
-        return {"alter_timezone.sql": macros__alter_timezone_sql}
+        return {
+            "alter_timezone.sql": macros__alter_timezone_sql,
+            "success_macro.sql": macros__success_macro_sql,
+        }
 
 
 class TestRetryNoPreviousRun(BaseTestRetry):
@@ -155,9 +165,10 @@ class TestRetryWarnError(BaseTestRetry):
 
 class TestRetryRunOperation(BaseTestRetry):
     def test_run_operation(self, project):
-        results = run_dbt(
+        results, log_output = run_dbt_and_capture(
             ["run-operation", "alter_timezone", "--args", "{timezone: abc}"], expect_pass=False
         )
+        assert "running a macro!" in log_output
 
         expected_statuses = {
             "macro.test.alter_timezone": RunStatus.Error,
@@ -165,7 +176,24 @@ class TestRetryRunOperation(BaseTestRetry):
 
         assert {n.unique_id: n.status for n in results.results} == expected_statuses
 
-        results = run_dbt(["retry"], expect_pass=False)
+        results, log_output = run_dbt_and_capture(["retry"], expect_pass=False)
+        assert "running a macro!" in log_output
+        assert {n.unique_id: n.status for n in results.results} == expected_statuses
+
+
+class TestRetrySuccessfulRunOperation(BaseTestRetry):
+    def test_run_operation(self, project):
+        results, log_output = run_dbt_and_capture(["run-operation", "success_macro"])
+        assert "running a macro!" in log_output
+
+        expected_statuses = {
+            "macro.test.success_macro": RunStatus.Success,
+        }
+
+        assert {n.unique_id: n.status for n in results.results} == expected_statuses
+
+        results, log_output = run_dbt_and_capture(["retry"])
+        assert "running a macro!" not in log_output
         assert {n.unique_id: n.status for n in results.results} == expected_statuses