Add refire_policy to PartitionedAssetTimetable for late/corrected events

airflow-core/docs/authoring-and-scheduling/assets.rst

@@ -723,6 +723,37 @@ including ``partition_key`` in the request body):
         "partition_key": "us|2026-03-10T09:00:00"
       }'
 
+Refiring on late or corrected events
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+By default, once a partition has triggered a Dag run, a later asset event for
+the same partition key does not affect that run — it starts a new, independent
+Dag run attempt instead. Set ``refire_policy`` on ``PartitionedAssetTimetable``
+to opt into different behavior for late-arriving or corrected events:
+
+.. code-block:: python
+
+    from airflow.sdk import DAG, PartitionedAssetTimetable
+
+    with DAG(
+        dag_id="clean_and_combine_player_stats",
+        schedule=PartitionedAssetTimetable(
+            assets=team_a_player_stats & team_b_player_stats,
+            refire_policy="rerun",
+        ),
+        catchup=False,
+    ):
+        ...
+
+* ``refire_policy=None`` (default) — no behavior change; each satisfied
+  partition triggers exactly one Dag run.
+* ``"rerun"`` — a late event for an already-fired partition clears and
+  re-queues the existing Dag run (its task instances are reset) instead of
+  leaving a duplicate, unsatisfiable Dag run behind.
+* ``"newrun"`` — a late event for an already-fired partition creates a new
+  Dag run, seeded with the prior run's other asset contributions for that
+  partition so the new run reflects the complete, corrected set of inputs.
+
 Rollup mappers
 ~~~~~~~~~~~~~~
 

airflow-core/src/airflow/assets/manager.py

@@ -695,6 +695,7 @@ def _queue_partitioned_dags(
                     rollup_fingerprint=fingerprint,
                     asset_id=asset_id,
                     allow_reuse=timetable.batch_asset_events,
+                    refire_policy=getattr(timetable, "refire_policy", None),
                     session=session,
                 )
                 log_record = PartitionedAssetKeyLog(
@@ -717,6 +718,7 @@ def _get_or_create_apdr(
         rollup_fingerprint: dict,
         asset_id: int,
         allow_reuse: bool,
+        refire_policy: str | None = None,
         session: Session,
     ) -> AssetPartitionDagRun:
         """
@@ -754,7 +756,7 @@ def _get_or_create_apdr(
         """
         with _lock_asset_model(session=session, asset_id=asset_id):
             latest_apdr: AssetPartitionDagRun | None = None
-            if allow_reuse:
+            if allow_reuse or refire_policy is not None:
                 latest_apdr = session.scalar(
                     select(AssetPartitionDagRun)
                     .where(
@@ -764,7 +766,7 @@ def _get_or_create_apdr(
                     .order_by(AssetPartitionDagRun.id.desc())
                     .limit(1)
                 )
-            if latest_apdr and latest_apdr.created_dag_run_id is None:
+            if allow_reuse and latest_apdr and latest_apdr.created_dag_run_id is None:
                 existing_partition_date = latest_apdr.partition_date
                 if existing_partition_date is None:
                     # No carried date yet; adopt the incoming one if present (no conflict
@@ -811,6 +813,44 @@ def _get_or_create_apdr(
                 target_dag.dag_id,
                 exc_info=True,
             )
+            # Refire: when the previous run for this partition already fired and the
+            # consumer opted into a refire_policy, carry forward the sibling assets'
+            # logged contributions so a single late/corrected event can re-satisfy the
+            # asset_condition. The scheduler then either re-runs the existing DagRun
+            # (refire_policy="rerun") or creates a new one (refire_policy="newrun").
+            if (
+                refire_policy is not None
+                and latest_apdr is not None
+                and latest_apdr.created_dag_run_id is not None
+            ):
+                prior_logs = session.scalars(
+                    select(PartitionedAssetKeyLog).where(
+                        PartitionedAssetKeyLog.asset_partition_dag_run_id == latest_apdr.id,
+                        PartitionedAssetKeyLog.asset_id != asset_id,
+                    )
+                ).all()
+                for prior in prior_logs:
+                    session.add(
+                        PartitionedAssetKeyLog(
+                            asset_id=prior.asset_id,
+                            asset_event_id=prior.asset_event_id,
+                            asset_partition_dag_run_id=apdr.id,
+                            source_partition_key=prior.source_partition_key,
+                            target_dag_id=prior.target_dag_id,
+                            target_partition_key=prior.target_partition_key,
+                        )
+                    )
+                if prior_logs:
+                    session.flush()
+                    cls.logger().info(
+                        "Refire: seeded %d sibling key-log row(s) from fired APDR %s into new APDR %s "
+                        "for key %s dag_id %s",
+                        len(prior_logs),
+                        latest_apdr.id,
+                        apdr.id,
+                        target_key,
+                        target_dag.dag_id,
+                    )
             return apdr
 
     @classmethod

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -115,7 +115,7 @@
 from airflow.models.pool import normalize_pool_name_for_stats
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.models.task_coordinates import LOOP_GATE_OPERATOR, TaskCoordinateResolver
-from airflow.models.taskinstance import TaskInstance
+from airflow.models.taskinstance import TaskInstance, clear_task_instances
 from airflow.models.team import Team
 from airflow.models.trigger import TRIGGER_FAIL_REPR, Trigger, TriggerFailureReason, handle_event_submit
 from airflow.observability.metrics import stats_utils
@@ -2610,6 +2610,45 @@ def _create_dagruns_for_partitioned_asset_dags(self, session: Session) -> set[st
                     dag_id=apdr.target_dag_id,
                     carried_partition_date=apdr.partition_date,
                 )
+            # Refire Semantics A ("rerun"): clear + re-queue the existing DagRun for this
+            # (dag, partition_key) instead of creating a new one, so a late/corrected
+            # event reprocesses in the same DagRun (attempts increment).
+            if getattr(timetable, "refire_policy", None) == "rerun":
+                existing_run = session.scalar(
+                    select(DagRun)
+                    .where(
+                        DagRun.dag_id == apdr.target_dag_id,
+                        DagRun.partition_key == apdr.partition_key,
+                    )
+                    .order_by(DagRun.id.desc())
+                    .limit(1)
+                )
+                if existing_run is not None:
+                    tis = session.scalars(
+                        select(TaskInstance).where(
+                            TaskInstance.dag_id == existing_run.dag_id,
+                            TaskInstance.run_id == existing_run.run_id,
+                        )
+                    ).all()
+                    clear_task_instances(list(tis), session, dag_run_state=DagRunState.QUEUED)
+                    existing_ids = {ev.id for ev in existing_run.consumed_asset_events}
+                    for ev in session.scalars(
+                        select(AssetEvent).where(
+                            PartitionedAssetKeyLog.asset_partition_dag_run_id == apdr.id,
+                            PartitionedAssetKeyLog.asset_event_id == AssetEvent.id,
+                        )
+                    ):
+                        if ev.id not in existing_ids:
+                            existing_run.consumed_asset_events.append(ev)
+                    apdr.created_dag_run_id = existing_run.id
+                    session.flush()
+                    self.log.info(
+                        "Refire(rerun): cleared and re-queued run %s for dag %s partition_key %s",
+                        existing_run.run_id,
+                        apdr.target_dag_id,
+                        apdr.partition_key,
+                    )
+                    continue
             dag_run = dag.create_dagrun(
                 run_id=DagRun.generate_run_id(
                     run_type=DagRunType.ASSET_TRIGGERED, logical_date=None, run_after=run_after

airflow-core/src/airflow/serialization/encoders.py

@@ -459,6 +459,7 @@ def _(self, timetable: PartitionedAssetTimetable) -> dict[str, Any]:
                 (encode_asset_like(asset), encode_partition_mapper(partition_mapper))
                 for asset, partition_mapper in timetable.partition_mapper_config.items()
             ],
+            "refire_policy": timetable.refire_policy,
         }
 
     BUILTIN_PARTITION_MAPPERS: dict[type, str] = {

airflow-core/src/airflow/timetables/simple.py

@@ -321,6 +321,10 @@ def summary(self) -> str:
         converter=attrs.converters.default_if_none(factory=dict),  # type: ignore[misc]
     )
     default_partition_mapper: PartitionMapper = DEFAULT_PARTITION_MAPPER
+    # Opt-in refire on late/corrected partition events: "rerun" (clear + re-run the
+    # existing DagRun, attempts increment) or "newrun" (a new DagRun per refire).
+    # None = fire once (default, no behavior change).
+    refire_policy: str | None = None
     batch_asset_events: bool = attrs.field(
         default=attrs.Factory(_get_default_batch_asset_events, takes_self=True)
     )
@@ -450,6 +454,7 @@ def serialize(self) -> dict[str, Any]:
                 for asset, partition_mapper in self.partition_mapper_config.items()
             ],
             "default_partition_mapper": encode_partition_mapper(self.default_partition_mapper),
+            "refire_policy": self.refire_policy,
         }
 
     @classmethod
@@ -468,5 +473,6 @@ def deserialize(cls, data: dict[str, Any]) -> PartitionedAssetTimetable:
                 decode_asset_like(ser_asest): decode_partition_mapper(ser_partition_mapper)
                 for ser_asest, ser_partition_mapper in partition_mapper_mappping_data
             },
+            refire_policy=data.get("refire_policy"),
         )
         return timetable

airflow-core/tests/unit/assets/test_manager.py

@@ -43,9 +43,11 @@
     DagScheduleAssetNameReference,
     DagScheduleAssetReference,
     DagScheduleAssetUriReference,
+    PartitionedAssetKeyLog,
 )
 from airflow.models.dag import DAG, DagModel
 from airflow.models.dagbundle import DagBundleModel
+from airflow.models.dagrun import DagRun
 from airflow.models.log import Log
 from airflow.models.team import Team
 from airflow.partition_mappers.identity import IdentityMapper
@@ -653,6 +655,131 @@ def test_get_or_create_apdr_allow_reuse(self, session, allow_reuse):
         assert session.scalar(select(func.count()).select_from(AssetPartitionDagRun)) == expected_count
         assert all(apdr.created_dag_run_id is None for apdr in apdrs)
 
+    @pytest.mark.usefixtures("testing_dag_bundle")
+    def test_get_or_create_apdr_refire_seeds_sibling_logs(self, session):
+        """A new APDR for an already-fired partition carries forward sibling key-log rows.
+
+        Once ``latest_apdr.created_dag_run_id`` is set, a plain ``_get_or_create_apdr`` call
+        always creates a fresh APDR holding only the arriving asset's key. With
+        ``refire_policy`` set, the prior APDR's key-log rows for *other* assets must be
+        copied onto the new APDR so the asset_condition can re-satisfy from this one event.
+        """
+        clear_db_apdr()
+        clear_db_pakl()
+        asset_a = AssetModel(uri="test://refire/a/", name="refire_a", group="asset")
+        asset_b = AssetModel(uri="test://refire/b/", name="refire_b", group="asset")
+        dag = DagModel(dag_id="refire_seed_dag", is_stale=False, bundle_name="testing")
+        session.add_all([asset_a, asset_b, dag])
+        session.flush()
+
+        fired_apdr = AssetManager._get_or_create_apdr(
+            target_key="key-1",
+            target_partition_date=None,
+            target_dag=dag,
+            rollup_fingerprint={},
+            asset_id=asset_a.id,
+            allow_reuse=True,
+            session=session,
+        )
+        sibling_event = AssetEvent(asset_id=asset_b.id, partition_key="key-1")
+        session.add(sibling_event)
+        session.flush()
+        session.add(
+            PartitionedAssetKeyLog(
+                asset_id=asset_b.id,
+                asset_event_id=sibling_event.id,
+                asset_partition_dag_run_id=fired_apdr.id,
+                source_partition_key="key-1",
+                target_dag_id=dag.dag_id,
+                target_partition_key="key-1",
+            )
+        )
+        # Simulate the scheduler having already materialized a DagRun for this APDR.
+        dag_run = DagRun(dag_id=dag.dag_id, run_id="refire_seed_run", run_type="asset_triggered")
+        session.add(dag_run)
+        session.flush()
+        fired_apdr.created_dag_run_id = dag_run.id
+        session.flush()
+
+        refired_apdr = AssetManager._get_or_create_apdr(
+            target_key="key-1",
+            target_partition_date=None,
+            target_dag=dag,
+            rollup_fingerprint={},
+            asset_id=asset_a.id,
+            allow_reuse=True,
+            refire_policy="rerun",
+            session=session,
+        )
+
+        assert refired_apdr.id != fired_apdr.id
+        assert refired_apdr.created_dag_run_id is None
+        seeded = session.scalars(
+            select(PartitionedAssetKeyLog).where(
+                PartitionedAssetKeyLog.asset_partition_dag_run_id == refired_apdr.id
+            )
+        ).all()
+        assert len(seeded) == 1
+        assert seeded[0].asset_id == asset_b.id
+        assert seeded[0].asset_event_id == sibling_event.id
+
+    @pytest.mark.usefixtures("testing_dag_bundle")
+    def test_get_or_create_apdr_no_refire_policy_does_not_seed_sibling_logs(self, session):
+        """Without a refire_policy, a new APDR for an already-fired partition stays empty."""
+        clear_db_apdr()
+        clear_db_pakl()
+        asset_a = AssetModel(uri="test://norefire/a/", name="norefire_a", group="asset")
+        asset_b = AssetModel(uri="test://norefire/b/", name="norefire_b", group="asset")
+        dag = DagModel(dag_id="norefire_seed_dag", is_stale=False, bundle_name="testing")
+        session.add_all([asset_a, asset_b, dag])
+        session.flush()
+
+        fired_apdr = AssetManager._get_or_create_apdr(
+            target_key="key-1",
+            target_partition_date=None,
+            target_dag=dag,
+            rollup_fingerprint={},
+            asset_id=asset_a.id,
+            allow_reuse=True,
+            session=session,
+        )
+        sibling_event = AssetEvent(asset_id=asset_b.id, partition_key="key-1")
+        session.add(sibling_event)
+        session.flush()
+        session.add(
+            PartitionedAssetKeyLog(
+                asset_id=asset_b.id,
+                asset_event_id=sibling_event.id,
+                asset_partition_dag_run_id=fired_apdr.id,
+                source_partition_key="key-1",
+                target_dag_id=dag.dag_id,
+                target_partition_key="key-1",
+            )
+        )
+        dag_run = DagRun(dag_id=dag.dag_id, run_id="norefire_seed_run", run_type="asset_triggered")
+        session.add(dag_run)
+        session.flush()
+        fired_apdr.created_dag_run_id = dag_run.id
+        session.flush()
+
+        refired_apdr = AssetManager._get_or_create_apdr(
+            target_key="key-1",
+            target_partition_date=None,
+            target_dag=dag,
+            rollup_fingerprint={},
+            asset_id=asset_a.id,
+            allow_reuse=True,
+            session=session,
+        )
+
+        assert refired_apdr.id != fired_apdr.id
+        seeded = session.scalars(
+            select(PartitionedAssetKeyLog).where(
+                PartitionedAssetKeyLog.asset_partition_dag_run_id == refired_apdr.id
+            )
+        ).all()
+        assert seeded == []
+
     @pytest.mark.need_serialized_dag
     @pytest.mark.usefixtures("testing_dag_bundle")
     def test_queue_partitioned_dags_stamps_rollup_fingerprint(self, session, dag_maker):