Support backfilling a subset of a Dag's tasks

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``c7f0a5d2e9b4`` (head) | ``76c46545c91e`` | ``3.4.0``         | Lower case team names.                                       |
+| ``8d1f4a2b6c37`` (head) | ``c7f0a5d2e9b4`` | ``3.4.0``         | Add selected_task_ids to backfill.                           |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``c7f0a5d2e9b4``        | ``76c46545c91e`` | ``3.4.0``         | Lower case team names.                                       |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``76c46545c91e``        | ``3c525f44bea8`` | ``3.4.0``         | Add new index for trigger.                                   |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

airflow-core/newsfragments/71978.feature.rst

@@ -0,0 +1 @@
+Support backfilling a subset of a Dag's tasks via a ``task_id_pattern`` on backfill creation: tasks whose id contains that value are scheduled, dependencies on unselected upstream tasks are ignored, and the resolved task ids are persisted so the selection stays reproducible; omitting it backfills the whole Dag as before.

airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py

@@ -37,6 +37,12 @@ class BackfillPostBody(StrictBaseModel):
     dag_run_conf: dict | None = None
     reprocess_behavior: ReprocessBehavior = ReprocessBehavior.NONE
     max_active_runs: int = 10
+    task_id_pattern: str | None = Field(
+        default=None,
+        description="Substring selecting a subset of tasks to backfill: tasks whose id contains this "
+        "value are scheduled and dependencies on unselected upstream tasks are ignored. "
+        "The request is rejected if it matches no task.",
+    )
     run_on_latest_version: bool | None = Field(
         default=None,
         description="Run on the latest bundle version of the Dag for each backfilled run. "
@@ -57,6 +63,7 @@ class BackfillResponse(BaseModel):
     is_paused: bool
     reprocess_behavior: ReprocessBehavior
     max_active_runs: int
+    selected_task_ids: list[str] | None
     created_at: datetime
     completed_at: datetime | None
     updated_at: datetime

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -2451,6 +2451,13 @@ components:
         max_active_runs:
           type: integer
           title: Max Active Runs
+        selected_task_ids:
+          anyOf:
+          - items:
+              type: string
+            type: array
+          - type: 'null'
+          title: Selected Task Ids
         created_at:
           type: string
           format: date-time
@@ -2478,6 +2485,7 @@ components:
       - is_paused
       - reprocess_behavior
       - max_active_runs
+      - selected_task_ids
       - created_at
       - completed_at
       - updated_at

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -11615,6 +11615,14 @@ components:
           type: integer
           title: Max Active Runs
           default: 10
+        task_id_pattern:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Task Id Pattern
+          description: 'Substring selecting a subset of tasks to backfill: tasks whose
+            id contains this value are scheduled and dependencies on unselected upstream
+            tasks are ignored. The request is rejected if it matches no task.'
         run_on_latest_version:
           anyOf:
           - type: boolean
@@ -11663,6 +11671,13 @@ components:
         max_active_runs:
           type: integer
           title: Max Active Runs
+        selected_task_ids:
+          anyOf:
+          - items:
+              type: string
+            type: array
+          - type: 'null'
+          title: Selected Task Ids
         created_at:
           type: string
           format: date-time
@@ -11690,6 +11705,7 @@ components:
       - is_paused
       - reprocess_behavior
       - max_active_runs
+      - selected_task_ids
       - created_at
       - completed_at
       - updated_at

airflow-core/src/airflow/api_fastapi/core_api/routes/public/backfills.py

@@ -59,6 +59,7 @@
     InvalidBackfillDirection,
     InvalidReprocessBehavior,
     NoBackfillRunsToCreate,
+    NoMatchingTasksForBackfill,
     _create_backfill,
     _do_dry_run,
 )
@@ -309,6 +310,7 @@ def create_backfill(
             triggering_user_name=user.get_display_name(),
             reprocess_behavior=backfill_request.reprocess_behavior,
             run_on_latest_version=resolved_run_on_latest,
+            task_id_pattern=backfill_request.task_id_pattern,
         )
         return BackfillResponse.model_validate(backfill_obj)
     except OperationalError as e:
@@ -338,6 +340,7 @@ def create_backfill(
         InvalidBackfillDateRange,
         InvalidBackfillConf,
         NoBackfillRunsToCreate,
+        NoMatchingTasksForBackfill,
     ) as e:
         raise RequestValidationError(str(e))
 
@@ -372,6 +375,7 @@ def create_backfill_dry_run(
             reprocess_behavior=body.reprocess_behavior,
             dag_run_conf=body.dag_run_conf,
             session=session,
+            task_id_pattern=body.task_id_pattern,
         )
         backfills = [
             DryRunBackfillResponse(
@@ -402,5 +406,6 @@ def create_backfill_dry_run(
         InvalidBackfillDate,
         InvalidBackfillDateRange,
         InvalidBackfillConf,
+        NoMatchingTasksForBackfill,
     ) as e:
         raise RequestValidationError(str(e))

airflow-core/src/airflow/migrations/versions/0132_3_4_0_add_selected_task_ids_to_backfill.py

@@ -0,0 +1,53 @@
+#
+# 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.
+
+"""
+Add selected_task_ids to backfill.
+
+Adds a nullable JSON column to the backfill table holding the resolved task
+ids a backfill is restricted to. A NULL value means the backfill covers the
+whole Dag (the existing behaviour).
+
+On PostgreSQL and MySQL 8+, adding a nullable column without a default is a
+metadata-only operation (no table rewrite).
+
+Revision ID: 8d1f4a2b6c37
+Revises: c7f0a5d2e9b4
+Create Date: 2026-08-22 00:00:00.000000
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "8d1f4a2b6c37"
+down_revision = "c7f0a5d2e9b4"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+    """Add selected_task_ids column to backfill table."""
+    op.add_column("backfill", sa.Column("selected_task_ids", sa.JSON(), nullable=True))
+
+
+def downgrade():
+    """Remove selected_task_ids column from backfill table."""
+    op.drop_column("backfill", "selected_task_ids")

airflow-core/src/airflow/models/backfill.py

@@ -135,6 +135,14 @@ class NoBackfillRunsToCreate(ValueError):
     """
 
 
+class NoMatchingTasksForBackfill(ValueError):
+    """
+    Raised when a backfill's task-id filter matches no tasks in the Dag.
+
+    :meta private:
+    """
+
+
 class UnknownActiveBackfills(AirflowException):
     """
     Raised when the quantity of active backfills cannot be determined.
@@ -178,6 +186,13 @@ class Backfill(Base):
         StringID(), nullable=False, default=ReprocessBehavior.NONE
     )
     max_active_runs: Mapped[int] = mapped_column(Integer, default=10, nullable=False)
+    selected_task_ids: Mapped[list | None] = mapped_column(sa.JSON(), nullable=True)
+    """
+    Resolved task ids this backfill is restricted to, or ``None`` for the whole Dag.
+
+    Holds the concrete task ids matched at creation time (not the pattern), so the
+    backfill remains reproducible even if the Dag later changes.
+    """
     created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=timezone.utcnow, nullable=False)
     completed_at: Mapped[datetime | None] = mapped_column(UtcDateTime, nullable=True)
     updated_at: Mapped[datetime] = mapped_column(
@@ -284,13 +299,32 @@ def _get_dag_run_no_create_reason(dr, reprocess_behavior: ReprocessBehavior) ->
     return non_create_reason
 
 
+def _resolve_backfill_task_ids(dag: SerializedDAG, task_id_pattern: str) -> list[str]:
+    """
+    Resolve a task-id substring filter to the sorted list of matching task ids in the Dag.
+
+    Matching is plain substring containment (a task is selected when its id contains
+    ``task_id_pattern``), mirroring ``SerializedDAG.partial_subset``. This deliberately
+    avoids compiling a user-supplied regular expression, which would be a ReDoS vector.
+
+    :raises NoMatchingTasksForBackfill: the filter matches no task in the Dag.
+    """
+    matched = sorted(task_id for task_id in dag.task_ids if task_id_pattern in task_id)
+    if not matched:
+        raise NoMatchingTasksForBackfill(
+            f"task_id_pattern {task_id_pattern!r} did not match any task in Dag {dag.dag_id}."
+        )
+    return matched
+
+
 def _validate_backfill_params(
     dag: SerializedDAG,
     reverse: bool,
     from_date: datetime,
     to_date: datetime,
     reprocess_behavior: ReprocessBehavior | None,
     dag_run_conf: dict | None = None,
+    selected_task_ids: list[str] | None = None,
 ) -> None:
 
     if from_date > to_date:
@@ -302,7 +336,9 @@ def _validate_backfill_params(
     if from_date >= current_time and to_date >= current_time:
         raise InvalidBackfillDate("Backfill cannot be executed for future dates.")
 
-    depends_on_past = any(x.depends_on_past for x in dag.tasks)
+    depends_on_past = any(
+        x.depends_on_past for x in dag.tasks if selected_task_ids is None or x.task_id in selected_task_ids
+    )
     if depends_on_past:
         if reverse is True:
             raise InvalidBackfillDirection(
@@ -329,6 +365,7 @@ def _do_dry_run(
     reprocess_behavior: ReprocessBehavior,
     session: Session,
     dag_run_conf: dict | None = None,
+    task_id_pattern: str | None = None,
 ) -> Iterable[DagRunInfo]:
     from airflow.models.serialized_dag import SerializedDagModel
 
@@ -344,13 +381,16 @@ def _do_dry_run(
     if dag.allowed_run_types is not None and DagRunType.BACKFILL_JOB not in dag.allowed_run_types:
         raise DagRunTypeNotAllowed(f"Dag with dag_id: '{dag_id}' does not allow backfill runs")
 
+    selected_task_ids = _resolve_backfill_task_ids(dag, task_id_pattern) if task_id_pattern else None
+
     _validate_backfill_params(
         dag,
         reverse,
         from_date,
         to_date,
         reprocess_behavior,
         dag_run_conf,
+        selected_task_ids,
     )
     dagrun_info_list = _get_info_list(
         dag=dag,
@@ -631,6 +671,7 @@ def _create_backfill(
     triggering_user_name: str | None,
     reprocess_behavior: ReprocessBehavior | None = None,
     run_on_latest_version: bool = False,
+    task_id_pattern: str | None = None,
 ) -> Backfill:
     from airflow.models import DagModel
     from airflow.models.serialized_dag import SerializedDagModel
@@ -668,13 +709,16 @@ def _create_backfill(
                 f"There can be only one running backfill per Dag."
             )
 
+        selected_task_ids = _resolve_backfill_task_ids(dag, task_id_pattern) if task_id_pattern else None
+
         _validate_backfill_params(
             dag,
             reverse,
             from_date,
             to_date,
             reprocess_behavior,
             dag_run_conf,
+            selected_task_ids,
         )
 
         dagrun_info_list = _get_info_list(
@@ -697,6 +741,7 @@ def _create_backfill(
             reprocess_behavior=reprocess_behavior,
             dag_model=dag,
             triggering_user_name=triggering_user_name,
+            selected_task_ids=selected_task_ids,
         )
         session.add(backfill)
         # Commit immediately so the backfill is visible to concurrent requests

airflow-core/src/airflow/models/dagbag.py

@@ -204,9 +204,19 @@ def _version_from_dag_run(dag_run: DagRun, *, session: Session) -> UUID | None:
         return dag_run.created_dag_version_id
 
     def get_dag_for_run(self, dag_run: DagRun, session: Session) -> SerializedDAG | None:
-        if version_id := self._version_from_dag_run(dag_run=dag_run, session=session):
-            return self._get_dag(version_id=version_id, session=session)
-        return None
+        if not (version_id := self._version_from_dag_run(dag_run=dag_run, session=session)):
+            return None
+        dag = self._get_dag(version_id=version_id, session=session)
+        if dag and dag_run.backfill_id and (backfill := dag_run.backfill) and backfill.selected_task_ids:
+            # Subset backfill: restrict the run to the selected tasks. partial_subset
+            # deep-copies, so the shared cached Dag is never mutated, and pruning the
+            # upstream/downstream references is what makes unselected upstreams ignored.
+            return dag.partial_subset(
+                backfill.selected_task_ids,
+                include_upstream=False,
+                include_downstream=False,
+            )
+        return dag
 
     def iter_all_latest_version_dags(self, *, session: Session) -> Generator[SerializedDAG, None, None]:
         """

airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -921,6 +921,18 @@ export const $BackfillPostBody = {
             title: 'Max Active Runs',
             default: 10
         },
+        task_id_pattern: {
+            anyOf: [
+                {
+                    type: 'string'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Task Id Pattern',
+            description: 'Substring selecting a subset of tasks to backfill: tasks whose id contains this value are scheduled and dependencies on unselected upstream tasks are ignored. The request is rejected if it matches no task.'
+        },
         run_on_latest_version: {
             anyOf: [
                 {
@@ -985,6 +997,20 @@ export const $BackfillResponse = {
             type: 'integer',
             title: 'Max Active Runs'
         },
+        selected_task_ids: {
+            anyOf: [
+                {
+                    items: {
+                        type: 'string'
+                    },
+                    type: 'array'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Selected Task Ids'
+        },
         created_at: {
             type: 'string',
             format: 'date-time',
@@ -1013,7 +1039,7 @@ export const $BackfillResponse = {
         }
     },
     type: 'object',
-    required: ['id', 'dag_id', 'from_date', 'to_date', 'dag_run_conf', 'is_paused', 'reprocess_behavior', 'max_active_runs', 'created_at', 'completed_at', 'updated_at', 'dag_display_name'],
+    required: ['id', 'dag_id', 'from_date', 'to_date', 'dag_run_conf', 'is_paused', 'reprocess_behavior', 'max_active_runs', 'selected_task_ids', 'created_at', 'completed_at', 'updated_at', 'dag_display_name'],
     title: 'BackfillResponse',
     description: 'Base serializer for Backfill.'
 } as const;