Add auto-clear of failed child tasks to TriggerDagRunOperator (#53402)

airflow-core/newsfragments/71730.feature.rst

@@ -0,0 +1 @@
+Add a new Execution API version (``2026-11-13``) that lets a running task clear only the failed tasks of an existing Dag run (and their downstream) instead of resetting the whole run. This backs the opt-in ``auto_clear_failed_tasks`` flag on ``TriggerDagRunOperator``: when a triggered child Dag run has already failed, re-running the parent task (for example on a retry) clears only the failed tasks so the child resumes from its point of failure while preserving already-succeeded tasks. ``reset_dag_run`` still takes precedence when both are set, and against an older core the operator raises ``NotImplementedError`` rather than silently falling back to a whole-run clear.

airflow-core/src/airflow/api_fastapi/execution_api/datamodels/dagrun.py

@@ -35,6 +35,12 @@ class TriggerDAGRunPayload(StrictBaseModel):
     note: str | None = None
 
 
+class ClearDagRunPayload(StrictBaseModel):
+    """Schema for Clear DAG Run API request."""
+
+    only_failed: bool = False
+
+
 class DagRunStateResponse(BaseModel):
     """Schema for DAG Run State response."""
 

airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py

@@ -21,7 +21,7 @@
 from typing import Annotated
 
 from cadwyn import VersionedAPIRouter
-from fastapi import HTTPException, Query, status
+from fastapi import Body, HTTPException, Query, status
 from sqlalchemy import func, select
 from sqlalchemy.exc import NoResultFound
 
@@ -30,7 +30,11 @@
 from airflow.api_fastapi.common.db.common import AsyncSessionDep, SessionDep
 from airflow.api_fastapi.common.types import UtcDateTime
 from airflow.api_fastapi.compat import HTTP_422_UNPROCESSABLE_CONTENT
-from airflow.api_fastapi.execution_api.datamodels.dagrun import DagRunStateResponse, TriggerDAGRunPayload
+from airflow.api_fastapi.execution_api.datamodels.dagrun import (
+    ClearDagRunPayload,
+    DagRunStateResponse,
+    TriggerDAGRunPayload,
+)
 from airflow.api_fastapi.execution_api.datamodels.taskinstance import DagRun
 from airflow.api_fastapi.execution_api.datamodels.token import TIToken
 from airflow.api_fastapi.execution_api.security import (
@@ -192,6 +196,7 @@ def clear_dag_run(
     run_id: str,
     session: SessionDep,
     dag_bag: DagBagDep,
+    payload: ClearDagRunPayload = Body(default_factory=ClearDagRunPayload),
 ) -> None:
     """Clear a Dag run."""
     dm = session.scalar(select(DagModel).where(~DagModel.is_stale, DagModel.dag_id == dag_id).limit(1))
@@ -221,7 +226,11 @@ def clear_dag_run(
     dag = get_dag_for_run(dag_bag, dag_run=dag_run, session=session)
 
     resolved_run_on_latest = resolve_run_on_latest_version(None, dag_id, session)
-    dag.clear(run_id=run_id, run_on_latest_version=resolved_run_on_latest)
+    dag.clear(
+        run_id=run_id,
+        only_failed=payload.only_failed,
+        run_on_latest_version=resolved_run_on_latest,
+    )
 
 
 @router.get(

airflow-core/src/airflow/api_fastapi/execution_api/versions/__init__.py

@@ -68,11 +68,18 @@
 from airflow.api_fastapi.execution_api.versions.v2026_10_30_xcom_params import (
     AddPreviousIterationToXComFilterParams,
 )
+from airflow.api_fastapi.execution_api.versions.v2026_11_13 import AddOnlyFailedToClearDagRunPayload
 from airflow.api_fastapi.execution_api.versions.v2027_02_28 import AddDagProcessorJobEndpoints
 
+# Minimum Execution API version that honours the ``only_failed`` clear scope. Clients (e.g. the
+# TriggerDagRunOperator failed-only path added in F1-T3) must gate on this and raise an explicit
+# error when negotiating against an older core, rather than silently clearing the whole run.
+MIN_VERSION_ONLY_FAILED_CLEAR = "2026-11-13"
+
 bundle = VersionBundle(
     HeadVersion(),
     Version("2027-02-28", AddDagProcessorJobEndpoints),
+    Version("2026-11-13", AddOnlyFailedToClearDagRunPayload),
     Version(
         "2026-10-30",
         AddArgBindingsToTIRunContext,

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2026_11_13.py

@@ -0,0 +1,35 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from cadwyn import VersionChange, schema
+
+from airflow.api_fastapi.execution_api.datamodels.dagrun import ClearDagRunPayload
+
+
+class AddOnlyFailedToClearDagRunPayload(VersionChange):
+    """Add the ``only_failed`` field to the clear-dag-run request payload."""
+
+    description = __doc__
+
+    # Request-body-only additive field: for older versions the field simply does not exist on
+    # ``ClearDagRunPayload`` (which forbids extra fields), so an older client omitting it clears the
+    # whole run, and one that sends it is rejected rather than silently downgraded to a whole-run clear.
+    instructions_to_migrate_to_previous_version = (
+        schema(ClearDagRunPayload).field("only_failed").didnt_exist,
+    )

airflow-core/tests/unit/api_fastapi/execution_api/datamodels/test_dagrun.py

@@ -0,0 +1,39 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from airflow.api_fastapi.execution_api.datamodels.dagrun import ClearDagRunPayload
+
+
+class TestClearDagRunPayload:
+    def test_only_failed_defaults_to_false(self):
+        """Omitting only_failed yields False so the route performs a whole-run clear."""
+        payload = ClearDagRunPayload()
+        assert payload.only_failed is False
+
+    @pytest.mark.parametrize("value", [True, False])
+    def test_only_failed_accepts_bool(self, value):
+        payload = ClearDagRunPayload(only_failed=value)
+        assert payload.only_failed is value
+
+    def test_only_failed_rejects_non_bool(self):
+        with pytest.raises(ValidationError):
+            ClearDagRunPayload(only_failed="not-a-bool")

airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_capabilities.py

@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from airflow.api_fastapi.execution_api.versions import (
+    MIN_VERSION_ONLY_FAILED_CLEAR,
+    bundle,
+)
+
+
+def test_min_version_only_failed_clear_is_registered():
+    """@AC-FR004-01 The advertised failed-only clear minimum maps to a registered API version."""
+    registered = {version.value for version in bundle.versions}
+    assert MIN_VERSION_ONLY_FAILED_CLEAR in registered
+
+
+def test_min_version_only_failed_clear_is_supported_by_head():
+    """@AC-FR004-02 The current head version is at least the failed-only clear minimum."""
+    # bundle.versions is ordered newest-first; the first entry is head/latest.
+    latest = bundle.versions[0].value
+    assert latest >= MIN_VERSION_ONLY_FAILED_CLEAR

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_11_13/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2026_11_13/test_dag_runs.py

@@ -0,0 +1,81 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+from unittest import mock
+
+import pytest
+from cadwyn import generate_versioned_models
+
+from airflow.api_fastapi.execution_api.datamodels.dagrun import ClearDagRunPayload
+from airflow.api_fastapi.execution_api.versions import bundle
+from airflow.serialization.definitions.dag import SerializedDAG
+from airflow.utils.state import DagRunState
+
+from tests_common.test_utils.db import clear_db_runs
+
+# The version that introduces failed-only clear and the last one that predates it.
+CURRENT_VERSION = "2026-11-13"
+PREVIOUS_VERSION = "2026-10-30"
+
+
+class TestClearDagRunPayloadVersioning:
+    """The ``only_failed`` field is versioned additively: present in the new version, absent before it."""
+
+    def test_field_present_in_current_version(self):
+        """@AC-FR004-01 The new version's ClearDagRunPayload exposes the additive only_failed field."""
+        versioned = generate_versioned_models(bundle)[CURRENT_VERSION][ClearDagRunPayload]
+        assert "only_failed" in versioned.model_fields
+
+    def test_field_negotiated_away_in_previous_version(self):
+        """@AC-FR004-02 The previous version negotiates only_failed away (field absent ⇒ whole-run)."""
+        versioned = generate_versioned_models(bundle)[PREVIOUS_VERSION][ClearDagRunPayload]
+        assert "only_failed" not in versioned.model_fields
+
+
+@pytest.mark.db_test
+class TestDagRunClearVersionNegotiation:
+    def setup_method(self):
+        clear_db_runs()
+
+    def teardown_method(self):
+        clear_db_runs()
+
+    @pytest.fixture
+    def old_ver_client(self, client):
+        """Client pinned to the version that predates the ``only_failed`` clear field."""
+        client.headers["Airflow-API-Version"] = PREVIOUS_VERSION
+        return client
+
+    def test_old_version_no_body_clears_whole_run(self, old_ver_client, session, dag_maker):
+        """@AC-FR004-01 An old-version client sending no body clears the whole run (only_failed defaults False)."""
+        from airflow.providers.standard.operators.empty import EmptyOperator
+
+        dag_id = "test_old_clear_no_body"
+        run_id = "test_run_id"
+        with dag_maker(dag_id=dag_id, session=session, serialized=True):
+            EmptyOperator(task_id="test_task")
+        dag_maker.create_dagrun(run_id=run_id, state=DagRunState.SUCCESS)
+        session.commit()
+
+        with mock.patch.object(SerializedDAG, "clear", autospec=True) as mock_clear:
+            response = old_ver_client.post(f"/execution/dag-runs/{dag_id}/{run_id}/clear")
+
+        assert response.status_code == 204
+        mock_clear.assert_called_once()
+        assert mock_clear.call_args.kwargs["only_failed"] is False

providers/standard/docs/changelog.rst

@@ -35,6 +35,16 @@
 Changelog
 ---------
 
+.. note::
+    ``TriggerDagRunOperator`` gains an opt-in ``auto_clear_failed_tasks`` flag (default ``False``).
+    When enabled and the triggered child Dag run already exists in a failed state, only its failed
+    tasks and their downstream are cleared and re-run on the next execution of the operator (for
+    example on a retry), preserving already-succeeded tasks instead of resetting the whole run. The
+    feature is synchronous-only in this version (``wait_for_completion=True``) and is not supported
+    with ``deferrable=True``. ``reset_dag_run`` still takes precedence when both are set. On Airflow
+    3.x the failed-only clear is delivered via the Execution API and requires a new enough core; the
+    operator raises ``NotImplementedError`` rather than silently falling back to a whole-run clear.
+
 1.20.1
 ......
 

providers/standard/docs/operators/trigger_dag_run.rst

@@ -29,3 +29,43 @@ Use the :class:`~airflow.providers.standard.operators.trigger_dagrun.TriggerDagR
     :dedent: 4
     :start-after: [START howto_operator_trigger_dagrun]
     :end-before: [END howto_operator_trigger_dagrun]
+
+Automatically clearing failed child tasks
+-----------------------------------------
+
+By default, when the triggered child Dag run already exists, ``TriggerDagRunOperator`` either
+raises, skips, or -- with ``reset_dag_run=True`` -- clears and re-runs the *whole* child run. The
+opt-in ``auto_clear_failed_tasks`` flag offers a narrower behavior: when the existing child run is in
+a failed state, only its failed tasks (and their downstream) are cleared and re-run, so
+already-succeeded upstream tasks are preserved and not re-executed.
+
+The flag is off by default. To enable it, set ``auto_clear_failed_tasks=True`` and run the operator
+synchronously (``wait_for_completion=True``). The failed-only clear happens on the next execution of
+the operator -- for example on a task retry -- so bound the attempts with ``retries``:
+
+.. code-block:: python
+
+    TriggerDagRunOperator(
+        task_id="run_child",
+        trigger_dag_id="child_dag",
+        wait_for_completion=True,  # sync mode (v1 scope)
+        retries=2,  # bounds the auto-clear + retry
+        auto_clear_failed_tasks=True,  # on retry, clear failed + downstream of the failed child run
+    )
+
+Caveats
+~~~~~~~
+
+* **Synchronous-only (v1).** ``auto_clear_failed_tasks`` applies when ``wait_for_completion=True`` and
+  is not supported with ``deferrable=True`` -- setting both raises ``ValueError``.
+* **Cosmetic window.** Until the operator re-executes and triggers the clear, the child run may still
+  be shown as ``failed`` in the UI.
+* **Task idempotency.** Previously-succeeded tasks are not re-run and the failed ones are, so the
+  triggered Dag's tasks should be idempotent; re-running a non-idempotent task may cause duplicate
+  side effects.
+* **Precedence with ``reset_dag_run``.** If both are set, ``reset_dag_run`` wins and the whole run is
+  cleared; at most one clear is performed.
+* **Airflow 3.x core-version requirement.** On Airflow 3.x the failed-only clear is delivered
+  server-side via the Execution API and requires a new enough core (Execution API version
+  ``2026-11-13`` or later). Against an older core the operator raises ``NotImplementedError`` rather
+  than silently falling back to a whole-run clear.