This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new f474e0ea6c7 [v3-3-test] Honor catchup for historical asset events in
asset-triggered Dags (#68749) (#69224)
f474e0ea6c7 is described below
commit f474e0ea6c765f71bb0341620faa92591d2cf960
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jul 3 19:21:12 2026 +0200
[v3-3-test] Honor catchup for historical asset events in asset-triggered
Dags (#68749) (#69224)
(cherry picked from commit ff10b2e3e19c9a6c1a262ebd32ef65b968b39424)
Co-authored-by: Shahar Epstein <[email protected]>
Co-authored-by: Jarek Potiuk <[email protected]>
---
airflow-core/newsfragments/68749.bugfix.rst | 1 +
.../src/airflow/jobs/scheduler_job_runner.py | 14 ++-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 123 +++++++++++++++++----
3 files changed, 113 insertions(+), 25 deletions(-)
diff --git a/airflow-core/newsfragments/68749.bugfix.rst
b/airflow-core/newsfragments/68749.bugfix.rst
new file mode 100644
index 00000000000..592b875b2f5
--- /dev/null
+++ b/airflow-core/newsfragments/68749.bugfix.rst
@@ -0,0 +1 @@
+Asset-triggered Dags now honor ``catchup`` for historical asset events. With
``catchup`` off (the default), a newly added asset-triggered Dag no longer
consumes events recorded before it started scheduling on those assets; its
first run is bounded at the moment the schedule reference was created instead
of reprocessing the entire backlog. With ``catchup`` on, the backlog is still
replayed (#39456).
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 2f94d480eb6..355f693db0a 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -2581,6 +2581,18 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
.cte()
)
+ # A first asset-triggered run has no previous run to floor the
event window. With
+ # catchup off, floor it at when the Dag started scheduling on its
assets so the
+ # backlog is skipped; with catchup on, only date.min applies and
the backlog replays.
+ event_window_floor: list[Any] = [cte.c.previous_dag_run_run_after]
+ if not dag.catchup:
+ event_window_floor.append(
+ select(func.min(DagScheduleAssetReference.created_at))
+ .where(DagScheduleAssetReference.dag_id == dag.dag_id)
+ .scalar_subquery()
+ )
+ event_window_floor.append(date.min)
+
asset_events = list(
session.scalars(
select(AssetEvent)
@@ -2598,7 +2610,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
),
),
AssetEvent.timestamp <= triggered_date,
- AssetEvent.timestamp >
func.coalesce(cte.c.previous_dag_run_run_after, date.min),
+ AssetEvent.timestamp >
func.coalesce(*event_window_floor),
)
.order_by(AssetEvent.timestamp.asc(), AssetEvent.id.asc())
)
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index e2d5977eb86..d7b08e662b1 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -65,6 +65,7 @@ from airflow.models.asset import (
AssetEvent,
AssetModel,
AssetPartitionDagRun,
+ DagScheduleAssetReference,
PartitionedAssetKeyLog,
)
from airflow.models.backfill import Backfill, BackfillDagRun,
ReprocessBehavior, _create_backfill
@@ -5501,51 +5502,59 @@ class TestSchedulerJob:
with dag_maker(dag_id="assets-1", start_date=timezone.utcnow(),
session=session):
BashOperator(task_id="task", bash_command="echo 1",
outlets=[asset1])
- dr = dag_maker.create_dagrun(
+ dr1 = dag_maker.create_dagrun(
run_id="run1",
logical_date=(DEFAULT_DATE + timedelta(days=100)),
data_interval=(DEFAULT_DATE + timedelta(days=10), DEFAULT_DATE +
timedelta(days=11)),
)
+ dr2 = dag_maker.create_dagrun(
+ run_id="run2",
+ logical_date=(DEFAULT_DATE + timedelta(days=101)),
+ data_interval=(DEFAULT_DATE + timedelta(days=5), DEFAULT_DATE +
timedelta(days=6)),
+ )
asset1_id = session.scalar(select(AssetModel.id).where(AssetModel.uri
== asset1.uri))
+ # Consumer Dags are created before the events, so the events fall
within their window.
+ with dag_maker(dag_id="assets-consumer-multiple", schedule=[asset1,
asset2]):
+ pass
+ dag2 = dag_maker.dag
+ with dag_maker(dag_id="assets-consumer-single", schedule=[asset1]):
+ pass
+ dag3 = dag_maker.dag
+
+ base = session.scalar(
+ select(DagScheduleAssetReference.created_at).where(
+ DagScheduleAssetReference.dag_id == dag3.dag_id
+ )
+ )
event1 = AssetEvent(
asset_id=asset1_id,
source_task_id="task",
- source_dag_id=dr.dag_id,
- source_run_id=dr.run_id,
+ source_dag_id=dr1.dag_id,
+ source_run_id=dr1.run_id,
source_map_index=-1,
+ timestamp=base + timedelta(seconds=1),
)
- session.add(event1)
-
- # Create a second event, creation time is more recent, but data
interval is older
- dr = dag_maker.create_dagrun(
- run_id="run2",
- logical_date=(DEFAULT_DATE + timedelta(days=101)),
- data_interval=(DEFAULT_DATE + timedelta(days=5), DEFAULT_DATE +
timedelta(days=6)),
- )
-
event2 = AssetEvent(
asset_id=asset1_id,
source_task_id="task",
- source_dag_id=dr.dag_id,
- source_run_id=dr.run_id,
+ source_dag_id=dr2.dag_id,
+ source_run_id=dr2.run_id,
source_map_index=-1,
+ timestamp=base + timedelta(seconds=2),
)
- session.add(event2)
-
- with dag_maker(dag_id="assets-consumer-multiple", schedule=[asset1,
asset2]):
- pass
- dag2 = dag_maker.dag
- with dag_maker(dag_id="assets-consumer-single", schedule=[asset1]):
- pass
- dag3 = dag_maker.dag
+ session.add_all([event1, event2])
session = dag_maker.session
session.add_all(
[
- AssetDagRunQueue(asset_id=asset1_id,
target_dag_id=dag2.dag_id),
- AssetDagRunQueue(asset_id=asset1_id,
target_dag_id=dag3.dag_id),
+ AssetDagRunQueue(
+ asset_id=asset1_id, target_dag_id=dag2.dag_id,
created_at=base + timedelta(hours=1)
+ ),
+ AssetDagRunQueue(
+ asset_id=asset1_id, target_dag_id=dag3.dag_id,
created_at=base + timedelta(hours=1)
+ ),
]
)
session.flush()
@@ -5591,6 +5600,72 @@ class TestSchedulerJob:
assert created_run.creating_job_id == scheduler_job.id
+ @pytest.mark.need_serialized_dag
+ @pytest.mark.parametrize(
+ ("catchup", "expects_old_event"),
+ [
+ pytest.param(False, False, id="catchup-off-ignores-backlog"),
+ pytest.param(True, True, id="catchup-on-consumes-backlog"),
+ ],
+ )
+ def test_new_asset_triggered_dag_backlog_gated_by_catchup(
+ self, catchup, expects_old_event, session, dag_maker
+ ):
+ """Reproduces #39456: catchup gates whether a new asset-triggered Dag
replays the
+ pre-creation backlog. With catchup off (the default) it only consumes
events after it
+ started scheduling on the asset; with catchup on it replays the full
history."""
+ asset = Asset(uri="test://asset-historical", name="hist_asset",
group="test_group")
+
+ # Producer Dag + run that the asset events are sourced from.
+ with dag_maker(dag_id="historical-producer",
start_date=timezone.utcnow(), session=session):
+ BashOperator(task_id="task", bash_command="echo 1",
outlets=[asset])
+ producer_run = dag_maker.create_dagrun(run_id="producer-run")
+
+ asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri
== asset.uri))
+
+ # Consumer Dag created now; its schedule reference's created_at is the
cut-off.
+ with dag_maker(dag_id="historical-consumer", schedule=[asset],
catchup=catchup):
+ pass
+ consumer_dag = dag_maker.dag
+ reference_created_at = session.scalar(
+ select(DagScheduleAssetReference.created_at).where(
+ DagScheduleAssetReference.dag_id == consumer_dag.dag_id
+ )
+ )
+
+ def _make_event(timestamp):
+ return AssetEvent(
+ asset_id=asset_id,
+ source_task_id="task",
+ source_dag_id=producer_run.dag_id,
+ source_run_id=producer_run.run_id,
+ source_map_index=-1,
+ timestamp=timestamp,
+ )
+
+ old_event = _make_event(reference_created_at - timedelta(days=1))
+ new_event = _make_event(reference_created_at + timedelta(seconds=1))
+ session.add_all([old_event, new_event])
+ # Trigger time after both events so neither is excluded by the upper
bound.
+ session.add(
+ AssetDagRunQueue(
+ asset_id=asset_id,
+ target_dag_id=consumer_dag.dag_id,
+ created_at=reference_created_at + timedelta(hours=1),
+ )
+ )
+ session.flush()
+
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job,
executors=[self.null_exec])
+ with create_session() as session:
+ self.job_runner._create_dagruns_for_dags(session, session)
+
+ created_run = session.scalars(select(DagRun).where(DagRun.dag_id ==
consumer_dag.dag_id)).one()
+ assert created_run.state == State.QUEUED
+ expected = {new_event.id} | ({old_event.id} if expects_old_event else
set())
+ assert {e.id for e in created_run.consumed_asset_events} == expected
+
@pytest.mark.need_serialized_dag
def test_create_dag_runs_asset_alias_with_asset_event_attached(self,
session, dag_maker):
"""
