This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 736fe690a95d81ac1701c7691c92c7ffb9a31390
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Thu Oct 29 10:00:53 2026 +0000
[v3-4-test] Avoid unbounded IN lists for queued asset events (#75943)
(#75948)
Co-authored-by: Tzu-ping Chung <[email protected]>
---
.../src/airflow/jobs/scheduler_job_runner.py | 158 +++++++-----
airflow-core/src/airflow/timetables/base.py | 4 +-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 270 +++++++++++++++++----
3 files changed, 324 insertions(+), 108 deletions(-)
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 7b8b2d53906..3817bcd450e 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -146,7 +146,7 @@ if TYPE_CHECKING:
from sqlalchemy.engine import CursorResult
from sqlalchemy.orm import Session
from sqlalchemy.orm.interfaces import LoaderOption
- from sqlalchemy.sql.elements import ColumnElement
+ from sqlalchemy.sql.elements import ColumnElement, SQLColumnExpression
from sqlalchemy.sql.selectable import Select, Subquery
from airflow._shared.logging.types import Logger
@@ -313,6 +313,16 @@ def _get_current_dr_task_concurrency(states:
Iterable[TaskInstanceState]) -> Sub
)
+def _consumed_by_dag(dag_id: str, event_id: SQLColumnExpression[int]) ->
ColumnElement[bool]:
+ """Check whether the asset event *event_id* has been consumed by a run of
the Dag."""
+ return (
+ select(association_table.c.event_id)
+ .join(DagRun, DagRun.id == association_table.c.dag_run_id)
+ .where(DagRun.dag_id == dag_id, association_table.c.event_id ==
event_id)
+ .exists()
+ )
+
+
class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
"""
SchedulerJobRunner runs for a specific time interval and schedules jobs
that are ready to run.
@@ -2777,18 +2787,17 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
)
continue
- consumed_asset_records: Sequence[AssetDagRunQueue] = ()
- gated_asset_events: list[AssetEvent] = []
if dag_model.timetable_asset_gated:
# The asset condition was evaluated without locks when this
Dag was
# selected; re-evaluate under ADRQ row locks so concurrent
schedulers
# cannot consume the same events twice. If the condition is
not (or no
# longer) satisfied, skip without touching the pending
schedule slot so
# a later loop retries it.
- gate = self._collect_gated_asset_events(dag=serdag,
session=session)
- if gate is None:
+ if (gate := self._collect_gated_asset_events(dag=serdag,
session=session)) is None:
continue
- consumed_asset_records, gated_asset_events = gate
+ gated_max_event_id, gated_asset_events = gate
+ else:
+ gated_max_event_id, gated_asset_events = None, ()
try:
next_info =
serdag.timetable.next_run_info_from_dag_model(dag_model=dag_model)
@@ -2823,10 +2832,13 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
session=session,
active_non_backfill_runs=active_runs_of_dags[dag_model.dag_id],
)
- if consumed_asset_records:
+ if gated_max_event_id is not None:
created_run.consumed_asset_events.extend(gated_asset_events)
+ session.flush() # Make consumed events visible to ADRQ
cleanup.
self._delete_consumed_asset_records(
- records=consumed_asset_records,
dag_id=dag_model.dag_id, session=session
+ dag_id=dag_model.dag_id,
+ max_event_id=gated_max_event_id,
+ session=session,
)
# Exceptions like ValueError, ParamValidationError, etc. are
raised by
@@ -2845,12 +2857,13 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
def _collect_gated_asset_events(
self, *, dag: SerializedDAG, session: Session
- ) -> tuple[Sequence[AssetDagRunQueue], list[AssetEvent]] | None:
+ ) -> tuple[int, Sequence[AssetEvent]] | None:
"""
Check an asset-gated Dag's asset condition and collect what a new run
consumes.
- Returns ``None`` when the condition is not satisfied by the queued
asset
- events, in which case no run should be created yet.
+ Returns the largest event ID among the locked ADRQ rows, and the events
+ the new run consumes. Returns ``None`` when the condition is not
satisfied
+ by the queued asset events, in which case no run should be created yet.
"""
records = self._lock_queued_asset_records(dag_id=dag.dag_id,
load_assets=True, session=session)
if not records:
@@ -2863,15 +2876,15 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
return None
if not ready:
return None
- asset_events = self._select_consumed_asset_events(
- dag=dag,
- records=records,
- session=session,
- )
+ max_event_id = max(record.asset_event_id for record in records)
+ asset_events = self._select_consumed_asset_events(dag=dag,
max_event_id=max_event_id, session=session)
if not asset_events:
- self._delete_consumed_asset_records(records=records,
dag_id=dag.dag_id, session=session)
+ # The select returns every queued event this Dag has not consumed
+ # yet; getting nothing back means all locked rows point to events
+ # that were already consumed. Delete those stale rows.
+ self._delete_consumed_asset_records(dag_id=dag.dag_id,
max_event_id=max_event_id, session=session)
return None
- return records, asset_events
+ return max_event_id, asset_events
def _lock_queued_asset_records(
self, *, dag_id: str, load_assets: bool, session: Session
@@ -2894,13 +2907,27 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
self,
*,
dag: SerializedDAG,
- records: Sequence[AssetDagRunQueue],
+ max_event_id: int,
limit: int | None = None,
session: Session,
- ) -> list[AssetEvent]:
- """Select unconsumed events referenced by a Dag's locked ADRQ rows."""
- referenced_event_ids = {record.asset_event_id for record in records}
- event_predicate: ColumnElement[bool] =
AssetEvent.id.in_(referenced_event_ids)
+ ) -> Sequence[AssetEvent]:
+ """
+ Select unconsumed events referenced by a Dag's locked ADRQ rows.
+
+ *max_event_id* is the largest event ID among the rows locked by this
+ scheduler, so events queued after the lock was taken are left for the
+ next scheduling loop. Concurrent schedulers can't consume this Dag's
rows
+ since the Dag's own row is locked in ``DagModel.dags_needing_dagruns``.
+ """
+ event_predicate: ColumnElement[bool] = (
+ select(AssetDagRunQueue.asset_event_id)
+ .where(
+ AssetDagRunQueue.target_dag_id == dag.dag_id,
+ AssetDagRunQueue.asset_event_id == AssetEvent.id,
+ AssetDagRunQueue.asset_event_id <= max_event_id,
+ )
+ .exists()
+ )
if dag.catchup:
event_predicate = or_(
event_predicate,
@@ -2913,42 +2940,45 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
AssetAliasModel.scheduled_dags.any(DagScheduleAssetAliasReference.dag_id ==
dag.dag_id)
),
)
- return list(
- session.scalars(
- select(AssetEvent)
- .where(
- event_predicate,
- ~(
- select(association_table.c.event_id)
- .join(DagRun, DagRun.id ==
association_table.c.dag_run_id)
- .where(
- DagRun.dag_id == dag.dag_id,
- association_table.c.event_id == AssetEvent.id,
- )
- .exists()
- ),
- )
- .order_by(AssetEvent.timestamp.asc(), AssetEvent.id.asc())
- .limit(limit)
+ return session.scalars(
+ select(AssetEvent)
+ .where(
+ event_predicate,
+ ~_consumed_by_dag(dag.dag_id, AssetEvent.id),
)
- )
+ .order_by(AssetEvent.timestamp.asc(), AssetEvent.id.asc())
+ .limit(limit)
+ ).all()
def _delete_consumed_asset_records(
- self, *, records: Sequence[AssetDagRunQueue], dag_id: str, session:
Session
+ self,
+ *,
+ dag_id: str,
+ max_event_id: int,
+ keep_event_id: int | None = None,
+ session: Session,
) -> None:
- # Delete only consumed ADRQ rows to avoid dropping newly queued events
- # (e.g. DagRun triggered by asset A while a new event for asset B
arrives).
- result = cast(
- "CursorResult",
- session.execute(
- delete(AssetDagRunQueue).where(
- tuple_(
- AssetDagRunQueue.target_dag_id,
- AssetDagRunQueue.asset_event_id,
- ).in_((record.target_dag_id, record.asset_event_id) for
record in records)
- )
- ),
+ """
+ Delete ADRQ rows whose events have been consumed by a run of the Dag.
+
+ Only consumed rows are deleted to avoid dropping newly queued events.
+ Rows are matched against the events consumed by the Dag's runs (i.e.
+ ``DagRun.consumed_asset_events``) with a subquery, so the caller must
+ flush newly consumed events before calling this.
+
+ :param max_event_id: Largest event ID among the ADRQ rows locked by
this
+ scheduler. Rows queued after the lock was taken are not touched.
+ :param keep_event_id: If given, the ADRQ row of this event is kept
even if
+ the event is consumed.
+ """
+ stmt = delete(AssetDagRunQueue).where(
+ AssetDagRunQueue.target_dag_id == dag_id,
+ AssetDagRunQueue.asset_event_id <= max_event_id,
+ _consumed_by_dag(dag_id, AssetDagRunQueue.asset_event_id),
)
+ if keep_event_id is not None:
+ stmt = stmt.where(AssetDagRunQueue.asset_event_id != keep_event_id)
+ result = cast("CursorResult", session.execute(stmt))
self.log.info(
"Deleted %d ADRQ rows for '%s'",
result.rowcount,
@@ -3007,9 +3037,10 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
if run_limit <= 0:
continue
+ max_event_id = max(record.asset_event_id for record in
queued_adrqs)
asset_events = self._select_consumed_asset_events(
dag=dag,
- records=queued_adrqs,
+ max_event_id=max_event_id,
limit=None if dag.timetable.batch_asset_events else run_limit
+ 1,
session=session,
)
@@ -3052,23 +3083,20 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
dag_run.run_id,
len(events),
)
+ session.flush() # Make consumed events visible to ADRQ
cleanup.
else:
self.log.info(
"No DagRun created for '%s' - asset events already
consumed or none found",
dag.dag_id,
)
- records_to_delete = queued_adrqs
- if has_more_events:
- consumed_ids = {event.id for event in asset_events}
- if dag.catchup:
- # Historical events may have no ADRQ. Keep a wakeup row
until that backlog drains.
- consumed_ids.discard(max(record.asset_event_id for record
in queued_adrqs))
- records_to_delete = [
- record for record in queued_adrqs if record.asset_event_id
in consumed_ids
- ]
+ # Dequeue rows for events both just consumed above and any stale
+ # ones. This runs even if no Dag run was created; otherwise those
+ # rows would keep the Dag selected in every scheduling loop.
self._delete_consumed_asset_records(
- records=records_to_delete,
dag_id=dag.dag_id,
+ max_event_id=max_event_id,
+ # Historical events may have no ADRQ. Keep a wakeup row until
that backlog drains.
+ keep_event_id=max_event_id if has_more_events and dag.catchup
else None,
session=session,
)
diff --git a/airflow-core/src/airflow/timetables/base.py
b/airflow-core/src/airflow/timetables/base.py
index a2325395a08..e1e5b8d6a0a 100644
--- a/airflow-core/src/airflow/timetables/base.py
+++ b/airflow-core/src/airflow/timetables/base.py
@@ -233,8 +233,8 @@ class Timetable(Protocol):
"""Whether queued asset events are consumed together in one Dag run."""
def group_asset_events(
- self, events: list[AssetEvent], triggered_date: DateTime
- ) -> Iterable[tuple[DateTime, list[AssetEvent]]]:
+ self, events: Sequence[AssetEvent], triggered_date: DateTime
+ ) -> Iterable[tuple[DateTime, Sequence[AssetEvent]]]:
"""Yield a run time and its consumed events for each asset-triggered
run."""
if not events:
return
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 37b440655ee..e808b722674 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -39,7 +39,7 @@ import pendulum
import psutil
import pytest
import time_machine
-from sqlalchemy import delete, func, inspect, select, update
+from sqlalchemy import delete, event as sa_event, func, inspect, select, update
from sqlalchemy.dialects import mysql
from sqlalchemy.orm import joinedload
@@ -7074,32 +7074,45 @@ class TestSchedulerJob:
assert session.scalars(select(DagRun).where(DagRun.dag_id ==
dag_model.dag_id)).one_or_none() is None
@pytest.mark.need_serialized_dag
+ @pytest.mark.parametrize(
+ ("locked_index", "expected_consumed_indexes",
"expected_queued_indexes"),
+ [
+ # The skipped row has a higher event ID than any locked row, so it
is
+ # outside the bound and left queued for the next scheduling loop.
+ pytest.param(0, {0}, {1}, id="skipped-row-newer"),
+ # The skipped row has a lower event ID than the locked row, so it
is
+ # within the bound, and its event is consumed and dequeued as well.
+ pytest.param(1, {0, 1}, set(), id="skipped-row-older"),
+ ],
+ )
def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows(
- self, session: Session, dag_maker
+ self,
+ session: Session,
+ dag_maker,
+ locked_index: int,
+ expected_consumed_indexes: set[int],
+ expected_queued_indexes: set[int],
):
- asset_1 = Asset("ready-to-trigger-a-Dag-run")
- asset_2 = Asset("should-still-exist-after-a-Dag-run-created")
- with dag_maker(dag_id="asset-consumer-delete-selected",
schedule=asset_1 | asset_2, session=session):
+ assets = [Asset("asset-delete-selected-1"),
Asset("asset-delete-selected-2")]
+ with dag_maker(
+ dag_id="asset-consumer-delete-selected", schedule=assets[0] |
assets[1], session=session
+ ):
pass
dag_model = dag_maker.dag_model
- asset_1_id = session.scalar(select(AssetModel.id).where(AssetModel.uri
== asset_1.name))
- asset_2_id = session.scalar(select(AssetModel.id).where(AssetModel.uri
== asset_2.name))
- event_1 = AssetEvent(asset_id=asset_1_id, timestamp=timezone.utcnow())
- event_2 = AssetEvent(asset_id=asset_2_id, timestamp=timezone.utcnow())
- session.add_all([event_1, event_2])
- session.flush()
+ asset_ids = [
+ session.scalar(select(AssetModel.id).where(AssetModel.uri ==
asset.name)) for asset in assets
+ ]
+ now = timezone.utcnow()
+ events = []
+ for i, asset_id in enumerate(asset_ids):
+ # Flush one by one so event IDs follow list order.
+ event = AssetEvent(asset_id=asset_id, timestamp=now +
timedelta(seconds=i))
+ session.add(event)
+ session.flush()
+ events.append(event)
session.add_all(
- [
- # The ADRQ that should trigger the Dag run creation
- AssetDagRunQueue(
- asset_id=asset_1_id, target_dag_id=dag_model.dag_id,
asset_event_id=event_1.id
- ),
- # The ADRQ that arrives after the Dag run creation but before
ADRQ clean up.
- # This situation is simulated by _lock_only_selected_asset
below.
- AssetDagRunQueue(
- asset_id=asset_2_id, target_dag_id=dag_model.dag_id,
asset_event_id=event_2.id
- ),
- ]
+ AssetDagRunQueue(asset_id=e.asset_id,
target_dag_id=dag_model.dag_id, asset_event_id=e.id)
+ for e in events
)
session.flush()
@@ -7107,8 +7120,8 @@ class TestSchedulerJob:
self.job_runner = SchedulerJobRunner(job=scheduler_job,
executors=[MockExecutor(do_update=False)])
def _lock_only_selected_asset(query, **_):
- # Simulate SKIP LOCKED behavior where this scheduler can only
consume one ADRQ row.
- return query.where(AssetDagRunQueue.asset_id == asset_1_id)
+ # Simulate SKIP LOCKED skipping the other row, e.g. held by
another transaction.
+ return query.where(AssetDagRunQueue.asset_id ==
asset_ids[locked_index])
with patch("airflow.jobs.scheduler_job_runner.with_row_locks",
side_effect=_lock_only_selected_asset):
self.job_runner._create_dag_runs_asset_triggered(
@@ -7116,23 +7129,14 @@ class TestSchedulerJob:
session=session,
)
- dr = session.scalars(select(DagRun).where(DagRun.dag_id ==
dag_model.dag_id)).one_or_none()
- assert dr is not None
-
- adrq_1 = session.scalars(
- select(AssetDagRunQueue).where(
- AssetDagRunQueue.target_dag_id == dag_model.dag_id,
- AssetDagRunQueue.asset_id == asset_1_id,
- )
- ).one_or_none()
- assert adrq_1 is None
- adrq_2 = session.scalars(
- select(AssetDagRunQueue).where(
- AssetDagRunQueue.target_dag_id == dag_model.dag_id,
- AssetDagRunQueue.asset_id == asset_2_id,
- )
- ).one_or_none()
- assert adrq_2 is not None
+ runs = session.scalars(select(DagRun).where(DagRun.dag_id ==
dag_model.dag_id)).all()
+ assert {e.id for run in runs for e in run.consumed_asset_events} == {
+ events[i].id for i in expected_consumed_indexes
+ }
+ queued_asset_ids = session.scalars(
+
select(AssetDagRunQueue.asset_id).where(AssetDagRunQueue.target_dag_id ==
dag_model.dag_id)
+ ).all()
+ assert set(queued_asset_ids) == {asset_ids[i] for i in
expected_queued_indexes}
@pytest.mark.need_serialized_dag
@pytest.mark.backend("postgres", "mysql")
@@ -12806,6 +12810,190 @@ def
test_create_dagruns_asset_and_time_late_arrival_uses_oldest_pending_logical_
)
[email protected]
+def _capture_max_bind_params(session: Session):
+ """Track the largest number of bind parameters sent in a single SQL
statement."""
+ counts: list[int] = [0]
+
+ def _listener(conn, cursor, statement, parameters, context, executemany):
+ if not executemany and parameters:
+ counts.append(len(parameters))
+
+ engine = session.get_bind()
+ sa_event.listen(engine, "before_cursor_execute", _listener)
+ try:
+ yield counts
+ finally:
+ sa_event.remove(engine, "before_cursor_execute", _listener)
+
+
+# Large enough that a literal IN list of queued events dwarfs any fixed-size
query.
+_MANY_QUEUED_EVENTS = 300
+
+
+def _queue_many_asset_events(
+ *,
+ dag_id: str,
+ asset_ids: list[int],
+ timestamp: datetime.datetime,
+ session: Session,
+) -> None:
+ events = [
+ AssetEvent(asset_id=asset_id, timestamp=timestamp)
+ for asset_id in asset_ids
+ for _ in range(_MANY_QUEUED_EVENTS)
+ ]
+ session.add_all(events)
+ session.flush()
+ session.add_all(
+ AssetDagRunQueue(asset_id=e.asset_id, target_dag_id=dag_id,
asset_event_id=e.id) for e in events
+ )
+ session.flush()
+
+
[email protected]_serialized_dag
[email protected]("batch_asset_events", [False, True])
+def test_create_dagruns_asset_triggered_many_queued_events_no_literal_in(
+ session: Session, dag_maker, batch_asset_events: bool
+):
+ """Queued events must not be sent to the database as a literal list
(#75865)."""
+ asset_1 = Asset(uri="test://many-events-1", name="many-events-1")
+ asset_2 = Asset(uri="test://many-events-2", name="many-events-2")
+ max_active_runs = 4 # Below the default per-loop run budget, so it is the
effective limit.
+ with dag_maker(
+ dag_id="many-events-triggered",
+ schedule=AssetTriggeredTimetable(assets=asset_1 & asset_2,
batch_asset_events=batch_asset_events),
+ max_active_runs=max_active_runs,
+ session=session,
+ ):
+ EmptyOperator(task_id="dummy_task")
+ dag_model = dag_maker.dag_model
+ asset_ids = list(
+
session.scalars(select(AssetModel.id).where(AssetModel.uri.in_([asset_1.uri,
asset_2.uri])))
+ )
+ _queue_many_asset_events(
+ dag_id=dag_model.dag_id,
+ asset_ids=asset_ids,
+ timestamp=timezone.utcnow(),
+ session=session,
+ )
+
+ runner = SchedulerJobRunner(job=Job(),
executors=[MockExecutor(do_update=False)])
+ with _capture_max_bind_params(session) as param_counts:
+ runner._create_dag_runs_asset_triggered(dag_models=[dag_model],
session=session)
+ session.flush()
+
+ assert max(param_counts) < _MANY_QUEUED_EVENTS
+ total = _MANY_QUEUED_EVENTS * len(asset_ids)
+ runs = session.scalars(select(DagRun).where(DagRun.dag_id ==
dag_model.dag_id)).all()
+ consumed = {e.id for run in runs for e in run.consumed_asset_events}
+ remaining = session.scalars(
+
select(AssetDagRunQueue.asset_event_id).where(AssetDagRunQueue.target_dag_id ==
dag_model.dag_id)
+ ).all()
+ if batch_asset_events:
+ assert len(runs) == 1
+ assert len(consumed) == total
+ else:
+ assert len(runs) == max_active_runs
+ assert len(consumed) == max_active_runs
+ assert consumed.isdisjoint(remaining)
+ assert len(remaining) == total - len(consumed)
+
+
[email protected]("disable_load_example")
[email protected]_serialized_dag
+def test_create_dagruns_asset_gated_many_queued_events_no_literal_in(session:
Session, dag_maker):
+ """Queued events must not be sent to the database as a literal list
(#75865)."""
+ asset_1 = Asset(uri="test://many-events-gated-1",
name="many-events-gated-1")
+ asset_2 = Asset(uri="test://many-events-gated-2",
name="many-events-gated-2")
+ logical_date = pendulum.datetime(2026, 3, 29, 17, tz="UTC")
+ with dag_maker(
+ dag_id="many-events-gated",
+ schedule=AssetAndTimeSchedule(
+ timetable=CronTriggerTimetable("* * * * *", timezone="UTC"),
+ assets=[asset_1, asset_2],
+ ),
+ session=session,
+ ):
+ EmptyOperator(task_id="dummy_task")
+ dag_model = dag_maker.dag_model
+ dag_model.next_dagrun = logical_date
+ dag_model.next_dagrun_data_interval = (logical_date, logical_date)
+ dag_model.next_dagrun_create_after = logical_date
+ asset_ids = list(
+
session.scalars(select(AssetModel.id).where(AssetModel.uri.in_([asset_1.uri,
asset_2.uri])))
+ )
+ _queue_many_asset_events(
+ dag_id=dag_model.dag_id,
+ asset_ids=asset_ids,
+ timestamp=logical_date,
+ session=session,
+ )
+
+ runner = SchedulerJobRunner(job=Job(), executors=[MockExecutor()])
+ with _capture_max_bind_params(session) as param_counts:
+ runner._create_dagruns_for_dags(cast("CommitProhibitorGuard",
session), session)
+ session.flush()
+
+ assert max(param_counts) < _MANY_QUEUED_EVENTS
+ dag_run = session.scalars(select(DagRun).where(DagRun.dag_id ==
dag_model.dag_id)).one()
+ assert len(dag_run.consumed_asset_events) == _MANY_QUEUED_EVENTS *
len(asset_ids)
+ assert (
+
session.scalar(select(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id ==
dag_model.dag_id))
+ is None
+ )
+
+
[email protected]("disable_load_example")
[email protected]_serialized_dag
+def test_create_dagruns_asset_gated_dequeues_already_consumed_events(session:
Session, dag_maker):
+ asset = Asset(uri="test://asset-gated-consumed",
name="asset-gated-consumed")
+ logical_date = pendulum.datetime(2026, 3, 29, 17, tz="UTC")
+ with dag_maker(
+ dag_id="asset-gated-consumed",
+ schedule=AssetAndTimeSchedule(
+ timetable=CronTriggerTimetable("* * * * *", timezone="UTC"),
+ assets=[asset],
+ ),
+ session=session,
+ ):
+ EmptyOperator(task_id="dummy_task")
+ dag_model = dag_maker.dag_model
+
+ # An event already consumed by an earlier run of this Dag...
+ asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri ==
asset.uri))
+ event = AssetEvent(asset_id=asset_id, timestamp=logical_date)
+ session.add(event)
+ session.flush()
+ prior_run = dag_maker.create_dagrun(
+ run_id="prior-consuming-run",
+ logical_date=logical_date - timedelta(hours=1),
+ data_interval=(logical_date - timedelta(hours=1), logical_date -
timedelta(hours=1)),
+ )
+ prior_run.consumed_asset_events.append(event)
+ # ... but still queued. This makes the asset condition look satisfied.
+ session.add(AssetDagRunQueue(asset_id=asset_id,
target_dag_id=dag_model.dag_id, asset_event_id=event.id))
+ dag_model.next_dagrun = logical_date
+ dag_model.next_dagrun_data_interval = (logical_date, logical_date)
+ dag_model.next_dagrun_create_after = logical_date
+ session.flush()
+
+ SchedulerJobRunner(job=Job(),
executors=[MockExecutor()])._create_dagruns_for_dags(
+ cast("CommitProhibitorGuard", session), session
+ )
+ session.flush()
+
+ # No run is created, the schedule slot is kept, and the stale row is
dequeued.
+ runs = session.scalars(select(DagRun).where(DagRun.dag_id ==
dag_model.dag_id)).all()
+ assert [run.run_id for run in runs] == ["prior-consuming-run"]
+ session.refresh(dag_model)
+ assert dag_model.next_dagrun == logical_date
+ assert (
+
session.scalar(select(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id ==
dag_model.dag_id))
+ is None
+ )
+
+
@time_machine.travel("2026-03-29 19:00:00+00:00")
@pytest.mark.usefixtures("disable_load_example")
@pytest.mark.need_serialized_dag
