Add row lock to ADRQ before Dag run creation#60773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Lee-W merged 6 commits into from
Mar 17, 2026
Merged

Add row lock to ADRQ before Dag run creation #60773

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
659ea0c
fix(ADRQ): lock ADRQs right before Dag run creationg to avoid race co…
Lee-W Mar 2, 2026
56ec448
style: ignore MockExecutor as it intends to be a mocked BaseExecutor
Lee-W Mar 2, 2026
6c474d1
refactor: address review comments
Lee-W Mar 5, 2026
b866aeb
refactor: sort by created_by so that we don't need to use max again
Lee-W Mar 5, 2026
e140c41
refactor: remove redundant triggered_date_by_dag usage
Lee-W Mar 16, 2026
51da4e1
fixup! refactor: remove redundant triggered_date_by_dag usage
Lee-W Mar 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 33 additions & 12 deletions airflow-core/src/airflow/jobs/scheduler_job_runner.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -1838,7 +1838,6 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio
if asset_triggered_dags:
self._create_dag_runs_asset_triggered(
dag_models=[d for d in asset_triggered_dags if d.dag_id not in partition_dag_ids],
triggered_date_by_dag=triggered_date_by_dag,
session=session,
)

Expand DownExpand Up@@ -2005,30 +2004,44 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -

def _create_dag_runs_asset_triggered(
self,
*,
dag_models: Collection[DagModel],
triggered_date_by_dag: dict[str, datetime],
session: Session,
) -> None:
"""For DAGs that are triggered by assets, create dag runs."""
triggered_dates: dict[str, DateTime] = {
dag_id: timezone.coerce_datetime(last_asset_event_time)
for dag_id, last_asset_event_time in triggered_date_by_dag.items()
}

"""For Dags that are triggered by assets, create Dag runs."""
for dag_model in dag_models:
dag = self._get_current_dag(dag_id=dag_model.dag_id, session=session)
if not dag:
self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
self.log.error("Dag '%s' not found in serialized_dag table", dag_model.dag_id)
continue

if not isinstance(dag.timetable, AssetTriggeredTimetable):
self.log.error(
"DAG '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
"Dag '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
dag_model.dag_id,
)
continue

triggered_date = triggered_dates[dag.dag_id]
queued_adrqs = session.scalars(
with_row_locks(
select(AssetDagRunQueue)
.where(AssetDagRunQueue.target_dag_id == dag.dag_id)
.order_by(AssetDagRunQueue.created_at.desc()),
of=AssetDagRunQueue,
skip_locked=True,
key_share=False,
session=session,
)
).all()
# If another scheduler already locked these ADRQ rows, SKIP LOCKED makes this scheduler skip them.
if not queued_adrqs:
self.log.debug(
"Skipping asset-triggered DagRun creation for Dag '%s'; no queued assets remain.",
dag.dag_id,
)
continue

triggered_date: DateTime = timezone.coerce_datetime(queued_adrqs[0].created_at)
cte = (
select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
.where(
Expand DownExpand Up@@ -2077,7 +2090,15 @@ def _create_dag_runs_asset_triggered(
)
Stats.incr("asset.triggered_dagruns")
dag_run.consumed_asset_events.extend(asset_events)
session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))

# 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).
adrq_pks = [(record.asset_id, record.target_dag_id) for record in queued_adrqs]
session.execute(
delete(AssetDagRunQueue).where(
tuple_(AssetDagRunQueue.asset_id, AssetDagRunQueue.target_dag_id).in_(adrq_pks)
)
)

def _lock_backfills(self, dag_runs: Collection[DagRun], session: Session) -> dict[int, Backfill]:
"""
Expand Down
89 changes: 87 additions & 2 deletions airflow-core/tests/unit/jobs/test_scheduler_job.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -285,9 +285,9 @@ def per_test(self) -> Generator:
def set_instance_attrs(self) -> Generator:
# Speed up some tests by not running the tasks, just look at what we
# enqueue!
self.null_exec: MockExecutor | None = MockExecutor()
self.null_exec: BaseExecutor = MockExecutor()
yield
self.null_exec = None
self.null_exec = None # type: ignore[assignment]

@pytest.fixture
def mock_executors(self):
Expand DownExpand Up@@ -4858,6 +4858,91 @@ def dict_from_obj(obj):

assert created_run.creating_job_id == scheduler_job.id

@pytest.mark.need_serialized_dag
def test_create_dag_runs_asset_triggered_skips_stale_triggered_date(self, session, dag_maker):
asset = Asset(uri="test://asset-for-stale-trigger-date", name="asset-for-stale-trigger-date")
with dag_maker(dag_id="asset-consumer-stale-trigger-date", schedule=[asset], session=session):
pass
dag_model = dag_maker.dag_model
asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri == asset.uri))

queued_at = timezone.utcnow()
session.add(AssetDagRunQueue(target_dag_id=dag_model.dag_id, asset_id=asset_id, created_at=queued_at))
session.flush()

# Simulate another scheduler consuming ADRQ rows after we computed triggered_date_by_dag.
session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_model.dag_id))
session.flush()

scheduler_job = Job()
self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])
self.job_runner._create_dag_runs_asset_triggered(
dag_models=[dag_model],
session=session,
)

# We do not create a new DagRun seems the ADRQ has already been consumed
assert session.scalars(select(DagRun).where(DagRun.dag_id == dag_model.dag_id)).one_or_none() is None

@pytest.mark.need_serialized_dag
def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows(
self, session: Session, dag_maker
):
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):
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))

session.add_all(
[
# The ADRQ that should triggers the Dag run creation
AssetDagRunQueue(
asset_id=asset_1_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
),
# The ADRQ that arrives after the Dag run creation but before ADRQ clean up
# This situation is simluarted by _lock_only_selected_asset below
AssetDagRunQueue(
asset_id=asset_2_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
),
]
)
session.flush()

scheduler_job = Job()
self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])

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)

with patch("airflow.jobs.scheduler_job_runner.with_row_locks", side_effect=_lock_only_selected_asset):
self.job_runner._create_dag_runs_asset_triggered(
dag_models=[dag_model],
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

@pytest.mark.need_serialized_dag
def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
"""
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { injectUserscript("// Add copy buttons to all 
 blocks\n(function() {\n function addCopyButtons() {\n document.querySelectorAll('pre code').forEach(function(codeBlock) {\n if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;\n codeBlock.parentElement.setAttribute('data-copy-added', 'true');\n \n var btn = document.createElement('button');\n btn.textContent = 'Copy';\n btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';\n btn.onmouseover = function() { this.style.opacity = '1'; };\n btn.onmouseout = function() { this.style.opacity = '0.7'; };\n btn.onclick = function() {\n navigator.clipboard.writeText(codeBlock.textContent).then(function() {\n btn.textContent = 'Copied!';\n setTimeout(function() { btn.textContent = 'Copy'; }, 1500);\n });\n };\n codeBlock.parentElement.style.position = 'relative';\n codeBlock.parentElement.appendChild(btn);\n });\n }\n \n addCopyButtons();\n \n // Re-run on dynamic content\n var observer = new MutationObserver(addCopyButtons);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Add Copy Buttons to Code Blocks");
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '

Skip to content
Add row lock to ADRQ before Dag run creation#60773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 6 commits into

from

Mar 17, 2026
 Merged


Add row lock to ADRQ before Dag run creation
#60773
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


6 commits


Select commit
Hold shift + click to select a range
659ea0c

fix(ADRQ): lock ADRQs right before Dag run creationg to avoid race co…


Lee-W Mar 2, 2026
56ec448

style: ignore MockExecutor as it intends to be a mocked BaseExecutor


Lee-W Mar 2, 2026
6c474d1

refactor: address review comments


Lee-W Mar 5, 2026
b866aeb

refactor: sort by created_by so that we don't need to use max again


Lee-W Mar 5, 2026
e140c41

refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026
51da4e1

fixup! refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


45 changes: 33 additions & 12 deletions
airflow-core/src/airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1838,7 +1838,6 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio
 if asset_triggered_dags:
 self._create_dag_runs_asset_triggered(
 dag_models=[d for d in asset_triggered_dags if d.dag_id not in partition_dag_ids],
 triggered_date_by_dag=triggered_date_by_dag,
 session=session,
 )

Expand DownExpand Up@@ -2005,30 +2004,44 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -

 def _create_dag_runs_asset_triggered(
 self,
 *,
 dag_models: Collection[DagModel],
 triggered_date_by_dag: dict[str, datetime],
 session: Session,
 ) -> None:
 """For DAGs that are triggered by assets, create dag runs."""
 triggered_dates: dict[str, DateTime] = {
 dag_id: timezone.coerce_datetime(last_asset_event_time)
 for dag_id, last_asset_event_time in triggered_date_by_dag.items()
 }

 """For Dags that are triggered by assets, create Dag runs."""
 for dag_model in dag_models:
 dag = self._get_current_dag(dag_id=dag_model.dag_id, session=session)
 if not dag:
 self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
 self.log.error("Dag '%s' not found in serialized_dag table", dag_model.dag_id)
 continue

 if not isinstance(dag.timetable, AssetTriggeredTimetable):
 self.log.error(
 "DAG '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 "Dag '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 dag_model.dag_id,
 )
 continue

 triggered_date = triggered_dates[dag.dag_id]
 queued_adrqs = session.scalars(
 with_row_locks(
 select(AssetDagRunQueue)
 .where(AssetDagRunQueue.target_dag_id == dag.dag_id)
 .order_by(AssetDagRunQueue.created_at.desc()),
 of=AssetDagRunQueue,
 skip_locked=True,
 key_share=False,
 session=session,
 )
 ).all()
 # If another scheduler already locked these ADRQ rows, SKIP LOCKED makes this scheduler skip them.
 if not queued_adrqs:
 self.log.debug(
 "Skipping asset-triggered DagRun creation for Dag '%s'; no queued assets remain.",
 dag.dag_id,
 )
 continue

 triggered_date: DateTime = timezone.coerce_datetime(queued_adrqs[0].created_at)
 cte = (
 select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
 .where(
Expand DownExpand Up@@ -2077,7 +2090,15 @@ def _create_dag_runs_asset_triggered(
 )
 Stats.incr("asset.triggered_dagruns")
 dag_run.consumed_asset_events.extend(asset_events)
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))

 # 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).
 adrq_pks = [(record.asset_id, record.target_dag_id) for record in queued_adrqs]
 session.execute(
 delete(AssetDagRunQueue).where(
 tuple_(AssetDagRunQueue.asset_id, AssetDagRunQueue.target_dag_id).in_(adrq_pks)
 )
 )

 def _lock_backfills(self, dag_runs: Collection[DagRun], session: Session) -> dict[int, Backfill]:
 """
Expand Down

89 changes: 87 additions & 2 deletions
airflow-core/tests/unit/jobs/test_scheduler_job.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -285,9 +285,9 @@ def per_test(self) -> Generator:
 def set_instance_attrs(self) -> Generator:
 # Speed up some tests by not running the tasks, just look at what we
 # enqueue!
 self.null_exec: MockExecutor | None = MockExecutor()
 self.null_exec: BaseExecutor = MockExecutor()
 yield
 self.null_exec = None
 self.null_exec = None # type: ignore[assignment]

 @pytest.fixture
 def mock_executors(self):
Expand DownExpand Up@@ -4858,6 +4858,91 @@ def dict_from_obj(obj):

 assert created_run.creating_job_id == scheduler_job.id

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_skips_stale_triggered_date(self, session, dag_maker):
 asset = Asset(uri="test://asset-for-stale-trigger-date", name="asset-for-stale-trigger-date")
 with dag_maker(dag_id="asset-consumer-stale-trigger-date", schedule=[asset], session=session):
 pass
 dag_model = dag_maker.dag_model
 asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri == asset.uri))

 queued_at = timezone.utcnow()
 session.add(AssetDagRunQueue(target_dag_id=dag_model.dag_id, asset_id=asset_id, created_at=queued_at))
 session.flush()

 # Simulate another scheduler consuming ADRQ rows after we computed triggered_date_by_dag.
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_model.dag_id))
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 session=session,
 )

 # We do not create a new DagRun seems the ADRQ has already been consumed
 assert session.scalars(select(DagRun).where(DagRun.dag_id == dag_model.dag_id)).one_or_none() is None

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows(
 self, session: Session, dag_maker
 ):
 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):
 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))

 session.add_all(
 [
 # The ADRQ that should triggers the Dag run creation
 AssetDagRunQueue(
 asset_id=asset_1_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 # The ADRQ that arrives after the Dag run creation but before ADRQ clean up
 # This situation is simluarted by _lock_only_selected_asset below
 AssetDagRunQueue(
 asset_id=asset_2_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 ]
 )
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])

 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)

 with patch("airflow.jobs.scheduler_job_runner.with_row_locks", side_effect=_lock_only_selected_asset):
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 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

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
 """
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Force GitHub README to respect dark mode\n(function() {\n var style = document.createElement('style');\n style.textContent = '\n .markdown-body {\n color-scheme: dark light;\n }\n .markdown-body pre { background: #161b22 !important; }\n .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }\n .markdown-body table th, .markdown-body table td { border-color: #30363d !important; }\n .markdown-body img { background: #0d1117; }\n .markdown-body blockquote { border-left-color: #8b949e; }\n .markdown-body hr { border-color: #30363d; }\n ';\n document.head.appendChild(style);\n})();", "GitHub Dark Mode README Fix");
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Add row lock to ADRQ before Dag run creation#60773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 6 commits into

from

Mar 17, 2026
 Merged


Add row lock to ADRQ before Dag run creation
#60773
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


6 commits


Select commit
Hold shift + click to select a range
659ea0c

fix(ADRQ): lock ADRQs right before Dag run creationg to avoid race co…


Lee-W Mar 2, 2026
56ec448

style: ignore MockExecutor as it intends to be a mocked BaseExecutor


Lee-W Mar 2, 2026
6c474d1

refactor: address review comments


Lee-W Mar 5, 2026
b866aeb

refactor: sort by created_by so that we don't need to use max again


Lee-W Mar 5, 2026
e140c41

refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026
51da4e1

fixup! refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


45 changes: 33 additions & 12 deletions
airflow-core/src/airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1838,7 +1838,6 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio
 if asset_triggered_dags:
 self._create_dag_runs_asset_triggered(
 dag_models=[d for d in asset_triggered_dags if d.dag_id not in partition_dag_ids],
 triggered_date_by_dag=triggered_date_by_dag,
 session=session,
 )

Expand DownExpand Up@@ -2005,30 +2004,44 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -

 def _create_dag_runs_asset_triggered(
 self,
 *,
 dag_models: Collection[DagModel],
 triggered_date_by_dag: dict[str, datetime],
 session: Session,
 ) -> None:
 """For DAGs that are triggered by assets, create dag runs."""
 triggered_dates: dict[str, DateTime] = {
 dag_id: timezone.coerce_datetime(last_asset_event_time)
 for dag_id, last_asset_event_time in triggered_date_by_dag.items()
 }

 """For Dags that are triggered by assets, create Dag runs."""
 for dag_model in dag_models:
 dag = self._get_current_dag(dag_id=dag_model.dag_id, session=session)
 if not dag:
 self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
 self.log.error("Dag '%s' not found in serialized_dag table", dag_model.dag_id)
 continue

 if not isinstance(dag.timetable, AssetTriggeredTimetable):
 self.log.error(
 "DAG '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 "Dag '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 dag_model.dag_id,
 )
 continue

 triggered_date = triggered_dates[dag.dag_id]
 queued_adrqs = session.scalars(
 with_row_locks(
 select(AssetDagRunQueue)
 .where(AssetDagRunQueue.target_dag_id == dag.dag_id)
 .order_by(AssetDagRunQueue.created_at.desc()),
 of=AssetDagRunQueue,
 skip_locked=True,
 key_share=False,
 session=session,
 )
 ).all()
 # If another scheduler already locked these ADRQ rows, SKIP LOCKED makes this scheduler skip them.
 if not queued_adrqs:
 self.log.debug(
 "Skipping asset-triggered DagRun creation for Dag '%s'; no queued assets remain.",
 dag.dag_id,
 )
 continue

 triggered_date: DateTime = timezone.coerce_datetime(queued_adrqs[0].created_at)
 cte = (
 select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
 .where(
Expand DownExpand Up@@ -2077,7 +2090,15 @@ def _create_dag_runs_asset_triggered(
 )
 Stats.incr("asset.triggered_dagruns")
 dag_run.consumed_asset_events.extend(asset_events)
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))

 # 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).
 adrq_pks = [(record.asset_id, record.target_dag_id) for record in queued_adrqs]
 session.execute(
 delete(AssetDagRunQueue).where(
 tuple_(AssetDagRunQueue.asset_id, AssetDagRunQueue.target_dag_id).in_(adrq_pks)
 )
 )

 def _lock_backfills(self, dag_runs: Collection[DagRun], session: Session) -> dict[int, Backfill]:
 """
Expand Down

89 changes: 87 additions & 2 deletions
airflow-core/tests/unit/jobs/test_scheduler_job.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -285,9 +285,9 @@ def per_test(self) -> Generator:
 def set_instance_attrs(self) -> Generator:
 # Speed up some tests by not running the tasks, just look at what we
 # enqueue!
 self.null_exec: MockExecutor | None = MockExecutor()
 self.null_exec: BaseExecutor = MockExecutor()
 yield
 self.null_exec = None
 self.null_exec = None # type: ignore[assignment]

 @pytest.fixture
 def mock_executors(self):
Expand DownExpand Up@@ -4858,6 +4858,91 @@ def dict_from_obj(obj):

 assert created_run.creating_job_id == scheduler_job.id

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_skips_stale_triggered_date(self, session, dag_maker):
 asset = Asset(uri="test://asset-for-stale-trigger-date", name="asset-for-stale-trigger-date")
 with dag_maker(dag_id="asset-consumer-stale-trigger-date", schedule=[asset], session=session):
 pass
 dag_model = dag_maker.dag_model
 asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri == asset.uri))

 queued_at = timezone.utcnow()
 session.add(AssetDagRunQueue(target_dag_id=dag_model.dag_id, asset_id=asset_id, created_at=queued_at))
 session.flush()

 # Simulate another scheduler consuming ADRQ rows after we computed triggered_date_by_dag.
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_model.dag_id))
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 session=session,
 )

 # We do not create a new DagRun seems the ADRQ has already been consumed
 assert session.scalars(select(DagRun).where(DagRun.dag_id == dag_model.dag_id)).one_or_none() is None

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows(
 self, session: Session, dag_maker
 ):
 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):
 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))

 session.add_all(
 [
 # The ADRQ that should triggers the Dag run creation
 AssetDagRunQueue(
 asset_id=asset_1_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 # The ADRQ that arrives after the Dag run creation but before ADRQ clean up
 # This situation is simluarted by _lock_only_selected_asset below
 AssetDagRunQueue(
 asset_id=asset_2_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 ]
 )
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])

 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)

 with patch("airflow.jobs.scheduler_job_runner.with_row_locks", side_effect=_lock_only_selected_asset):
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 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

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
 """
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Highlight search terms from Google/DuckDuckGo/Bing referrer\n(function() {\n var ref = document.referrer;\n var terms = [];\n \n if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {\n var url = new URL(ref);\n var q = url.searchParams.get('q') || url.searchParams.get('p');\n if (q) {\n terms = q.split(/\\s+/).filter(function(t) { return t.length > 2; });\n }\n }\n \n if (terms.length === 0) return;\n \n var style = document.createElement('style');\n style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';\n document.head.appendChild(style);\n \n function highlight(node) {\n if (node.nodeType === 3) { // text node\n var text = node.textContent;\n var found = false;\n terms.forEach(function(term) {\n var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\\]\\\\]/g, '\\\\') + ')', 'gi');\n if (regex.test(text)) {\n found = true;\n var frag = document.createDocumentFragment();\n var parts = text.split(regex);\n parts.forEach(function(part, i) {\n if (i % 2 === 0) {\n frag.appendChild(document.createTextNode(part));\n } else {\n var span = document.createElement('span');\n span.className = 'userscript-highlight';\n span.textContent = part;\n frag.appendChild(span);\n }\n });\n node.parentNode.replaceChild(frag, node);\n }\n });\n } else if (node.nodeType === 1 && node.childNodes) { // element\n var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];\n if (!skipTags.includes(node.tagName)) {\n Array.from(node.childNodes).forEach(highlight);\n }\n }\n }\n \n highlight(document.body);\n \n // Re-highlight on dynamic content\n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1 || node.nodeType === 3) highlight(node);\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Highlight Search Terms");
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Add row lock to ADRQ before Dag run creation#60773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 6 commits into

from

Mar 17, 2026
 Merged


Add row lock to ADRQ before Dag run creation
#60773
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


6 commits


Select commit
Hold shift + click to select a range
659ea0c

fix(ADRQ): lock ADRQs right before Dag run creationg to avoid race co…


Lee-W Mar 2, 2026
56ec448

style: ignore MockExecutor as it intends to be a mocked BaseExecutor


Lee-W Mar 2, 2026
6c474d1

refactor: address review comments


Lee-W Mar 5, 2026
b866aeb

refactor: sort by created_by so that we don't need to use max again


Lee-W Mar 5, 2026
e140c41

refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026
51da4e1

fixup! refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


45 changes: 33 additions & 12 deletions
airflow-core/src/airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1838,7 +1838,6 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio
 if asset_triggered_dags:
 self._create_dag_runs_asset_triggered(
 dag_models=[d for d in asset_triggered_dags if d.dag_id not in partition_dag_ids],
 triggered_date_by_dag=triggered_date_by_dag,
 session=session,
 )

Expand DownExpand Up@@ -2005,30 +2004,44 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -

 def _create_dag_runs_asset_triggered(
 self,
 *,
 dag_models: Collection[DagModel],
 triggered_date_by_dag: dict[str, datetime],
 session: Session,
 ) -> None:
 """For DAGs that are triggered by assets, create dag runs."""
 triggered_dates: dict[str, DateTime] = {
 dag_id: timezone.coerce_datetime(last_asset_event_time)
 for dag_id, last_asset_event_time in triggered_date_by_dag.items()
 }

 """For Dags that are triggered by assets, create Dag runs."""
 for dag_model in dag_models:
 dag = self._get_current_dag(dag_id=dag_model.dag_id, session=session)
 if not dag:
 self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
 self.log.error("Dag '%s' not found in serialized_dag table", dag_model.dag_id)
 continue

 if not isinstance(dag.timetable, AssetTriggeredTimetable):
 self.log.error(
 "DAG '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 "Dag '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 dag_model.dag_id,
 )
 continue

 triggered_date = triggered_dates[dag.dag_id]
 queued_adrqs = session.scalars(
 with_row_locks(
 select(AssetDagRunQueue)
 .where(AssetDagRunQueue.target_dag_id == dag.dag_id)
 .order_by(AssetDagRunQueue.created_at.desc()),
 of=AssetDagRunQueue,
 skip_locked=True,
 key_share=False,
 session=session,
 )
 ).all()
 # If another scheduler already locked these ADRQ rows, SKIP LOCKED makes this scheduler skip them.
 if not queued_adrqs:
 self.log.debug(
 "Skipping asset-triggered DagRun creation for Dag '%s'; no queued assets remain.",
 dag.dag_id,
 )
 continue

 triggered_date: DateTime = timezone.coerce_datetime(queued_adrqs[0].created_at)
 cte = (
 select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
 .where(
Expand DownExpand Up@@ -2077,7 +2090,15 @@ def _create_dag_runs_asset_triggered(
 )
 Stats.incr("asset.triggered_dagruns")
 dag_run.consumed_asset_events.extend(asset_events)
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))

 # 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).
 adrq_pks = [(record.asset_id, record.target_dag_id) for record in queued_adrqs]
 session.execute(
 delete(AssetDagRunQueue).where(
 tuple_(AssetDagRunQueue.asset_id, AssetDagRunQueue.target_dag_id).in_(adrq_pks)
 )
 )

 def _lock_backfills(self, dag_runs: Collection[DagRun], session: Session) -> dict[int, Backfill]:
 """
Expand Down

89 changes: 87 additions & 2 deletions
airflow-core/tests/unit/jobs/test_scheduler_job.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -285,9 +285,9 @@ def per_test(self) -> Generator:
 def set_instance_attrs(self) -> Generator:
 # Speed up some tests by not running the tasks, just look at what we
 # enqueue!
 self.null_exec: MockExecutor | None = MockExecutor()
 self.null_exec: BaseExecutor = MockExecutor()
 yield
 self.null_exec = None
 self.null_exec = None # type: ignore[assignment]

 @pytest.fixture
 def mock_executors(self):
Expand DownExpand Up@@ -4858,6 +4858,91 @@ def dict_from_obj(obj):

 assert created_run.creating_job_id == scheduler_job.id

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_skips_stale_triggered_date(self, session, dag_maker):
 asset = Asset(uri="test://asset-for-stale-trigger-date", name="asset-for-stale-trigger-date")
 with dag_maker(dag_id="asset-consumer-stale-trigger-date", schedule=[asset], session=session):
 pass
 dag_model = dag_maker.dag_model
 asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri == asset.uri))

 queued_at = timezone.utcnow()
 session.add(AssetDagRunQueue(target_dag_id=dag_model.dag_id, asset_id=asset_id, created_at=queued_at))
 session.flush()

 # Simulate another scheduler consuming ADRQ rows after we computed triggered_date_by_dag.
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_model.dag_id))
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 session=session,
 )

 # We do not create a new DagRun seems the ADRQ has already been consumed
 assert session.scalars(select(DagRun).where(DagRun.dag_id == dag_model.dag_id)).one_or_none() is None

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows(
 self, session: Session, dag_maker
 ):
 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):
 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))

 session.add_all(
 [
 # The ADRQ that should triggers the Dag run creation
 AssetDagRunQueue(
 asset_id=asset_1_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 # The ADRQ that arrives after the Dag run creation but before ADRQ clean up
 # This situation is simluarted by _lock_only_selected_asset below
 AssetDagRunQueue(
 asset_id=asset_2_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 ]
 )
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])

 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)

 with patch("airflow.jobs.scheduler_job_runner.with_row_locks", side_effect=_lock_only_selected_asset):
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 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

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
 """
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Strip utm_, fbclid, gclid, etc. from all links on page\n(function() {\n var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',\n 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',\n 'ref', 'ref_src', 'source', 'medium', 'campaign'];\n \n function cleanUrl(url) {\n try {\n var u = new URL(url, window.location.origin);\n var changed = false;\n trackingParams.forEach(function(p) {\n if (u.searchParams.has(p)) {\n u.searchParams.delete(p);\n changed = true;\n }\n });\n return changed ? u.toString() : url;\n } catch (e) {\n return url;\n }\n }\n \n function cleanLinks() {\n document.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n \n cleanLinks();\n \n var observer = new MutationObserver(function(mutations) {\n mutations.forEach(function(m) {\n m.addedNodes.forEach(function(node) {\n if (node.nodeType === 1) {\n if (node.tagName === 'A') cleanLinks();\n node.querySelectorAll('a[href]').forEach(function(a) {\n var clean = cleanUrl(a.href);\n if (clean !== a.href) a.href = clean;\n });\n }\n });\n });\n });\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "Remove Tracking Parameters from Links");
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '

Skip to content
Add row lock to ADRQ before Dag run creation#60773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 6 commits into

from

Mar 17, 2026
 Merged


Add row lock to ADRQ before Dag run creation
#60773
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


6 commits


Select commit
Hold shift + click to select a range
659ea0c

fix(ADRQ): lock ADRQs right before Dag run creationg to avoid race co…


Lee-W Mar 2, 2026
56ec448

style: ignore MockExecutor as it intends to be a mocked BaseExecutor


Lee-W Mar 2, 2026
6c474d1

refactor: address review comments


Lee-W Mar 5, 2026
b866aeb

refactor: sort by created_by so that we don't need to use max again


Lee-W Mar 5, 2026
e140c41

refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026
51da4e1

fixup! refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


45 changes: 33 additions & 12 deletions
airflow-core/src/airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1838,7 +1838,6 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio
 if asset_triggered_dags:
 self._create_dag_runs_asset_triggered(
 dag_models=[d for d in asset_triggered_dags if d.dag_id not in partition_dag_ids],
 triggered_date_by_dag=triggered_date_by_dag,
 session=session,
 )

Expand DownExpand Up@@ -2005,30 +2004,44 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -

 def _create_dag_runs_asset_triggered(
 self,
 *,
 dag_models: Collection[DagModel],
 triggered_date_by_dag: dict[str, datetime],
 session: Session,
 ) -> None:
 """For DAGs that are triggered by assets, create dag runs."""
 triggered_dates: dict[str, DateTime] = {
 dag_id: timezone.coerce_datetime(last_asset_event_time)
 for dag_id, last_asset_event_time in triggered_date_by_dag.items()
 }

 """For Dags that are triggered by assets, create Dag runs."""
 for dag_model in dag_models:
 dag = self._get_current_dag(dag_id=dag_model.dag_id, session=session)
 if not dag:
 self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
 self.log.error("Dag '%s' not found in serialized_dag table", dag_model.dag_id)
 continue

 if not isinstance(dag.timetable, AssetTriggeredTimetable):
 self.log.error(
 "DAG '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 "Dag '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 dag_model.dag_id,
 )
 continue

 triggered_date = triggered_dates[dag.dag_id]
 queued_adrqs = session.scalars(
 with_row_locks(
 select(AssetDagRunQueue)
 .where(AssetDagRunQueue.target_dag_id == dag.dag_id)
 .order_by(AssetDagRunQueue.created_at.desc()),
 of=AssetDagRunQueue,
 skip_locked=True,
 key_share=False,
 session=session,
 )
 ).all()
 # If another scheduler already locked these ADRQ rows, SKIP LOCKED makes this scheduler skip them.
 if not queued_adrqs:
 self.log.debug(
 "Skipping asset-triggered DagRun creation for Dag '%s'; no queued assets remain.",
 dag.dag_id,
 )
 continue

 triggered_date: DateTime = timezone.coerce_datetime(queued_adrqs[0].created_at)
 cte = (
 select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
 .where(
Expand DownExpand Up@@ -2077,7 +2090,15 @@ def _create_dag_runs_asset_triggered(
 )
 Stats.incr("asset.triggered_dagruns")
 dag_run.consumed_asset_events.extend(asset_events)
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))

 # 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).
 adrq_pks = [(record.asset_id, record.target_dag_id) for record in queued_adrqs]
 session.execute(
 delete(AssetDagRunQueue).where(
 tuple_(AssetDagRunQueue.asset_id, AssetDagRunQueue.target_dag_id).in_(adrq_pks)
 )
 )

 def _lock_backfills(self, dag_runs: Collection[DagRun], session: Session) -> dict[int, Backfill]:
 """
Expand Down

89 changes: 87 additions & 2 deletions
airflow-core/tests/unit/jobs/test_scheduler_job.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -285,9 +285,9 @@ def per_test(self) -> Generator:
 def set_instance_attrs(self) -> Generator:
 # Speed up some tests by not running the tasks, just look at what we
 # enqueue!
 self.null_exec: MockExecutor | None = MockExecutor()
 self.null_exec: BaseExecutor = MockExecutor()
 yield
 self.null_exec = None
 self.null_exec = None # type: ignore[assignment]

 @pytest.fixture
 def mock_executors(self):
Expand DownExpand Up@@ -4858,6 +4858,91 @@ def dict_from_obj(obj):

 assert created_run.creating_job_id == scheduler_job.id

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_skips_stale_triggered_date(self, session, dag_maker):
 asset = Asset(uri="test://asset-for-stale-trigger-date", name="asset-for-stale-trigger-date")
 with dag_maker(dag_id="asset-consumer-stale-trigger-date", schedule=[asset], session=session):
 pass
 dag_model = dag_maker.dag_model
 asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri == asset.uri))

 queued_at = timezone.utcnow()
 session.add(AssetDagRunQueue(target_dag_id=dag_model.dag_id, asset_id=asset_id, created_at=queued_at))
 session.flush()

 # Simulate another scheduler consuming ADRQ rows after we computed triggered_date_by_dag.
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_model.dag_id))
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 session=session,
 )

 # We do not create a new DagRun seems the ADRQ has already been consumed
 assert session.scalars(select(DagRun).where(DagRun.dag_id == dag_model.dag_id)).one_or_none() is None

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows(
 self, session: Session, dag_maker
 ):
 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):
 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))

 session.add_all(
 [
 # The ADRQ that should triggers the Dag run creation
 AssetDagRunQueue(
 asset_id=asset_1_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 # The ADRQ that arrives after the Dag run creation but before ADRQ clean up
 # This situation is simluarted by _lock_only_selected_asset below
 AssetDagRunQueue(
 asset_id=asset_2_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 ]
 )
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])

 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)

 with patch("airflow.jobs.scheduler_job_runner.with_row_locks", side_effect=_lock_only_selected_asset):
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 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

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
 """
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Auto-enable theater mode on YouTube\n(function() {\n function tryTheater() {\n var btn = document.querySelector('button[aria-label=\"Theater mode\"], ytd-player #player button[title=\"Theater mode\"]');\n if (btn && !btn.classList.contains('activated')) {\n btn.click();\n }\n }\n \n // Try immediately\n tryTheater();\n \n // Try after navigation (SPA)\n var lastUrl = location.href;\n setInterval(function() {\n if (location.href !== lastUrl) {\n lastUrl = location.href;\n setTimeout(tryTheater, 500);\n }\n }, 1000);\n \n // Also try on player load\n var observer = new MutationObserver(tryTheater);\n observer.observe(document.body, { childList: true, subtree: true });\n})();", "YouTube Theater Mode Default");
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Add row lock to ADRQ before Dag run creation#60773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 6 commits into

from

Mar 17, 2026
 Merged


Add row lock to ADRQ before Dag run creation
#60773
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


6 commits


Select commit
Hold shift + click to select a range
659ea0c

fix(ADRQ): lock ADRQs right before Dag run creationg to avoid race co…


Lee-W Mar 2, 2026
56ec448

style: ignore MockExecutor as it intends to be a mocked BaseExecutor


Lee-W Mar 2, 2026
6c474d1

refactor: address review comments


Lee-W Mar 5, 2026
b866aeb

refactor: sort by created_by so that we don't need to use max again


Lee-W Mar 5, 2026
e140c41

refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026
51da4e1

fixup! refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


45 changes: 33 additions & 12 deletions
airflow-core/src/airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1838,7 +1838,6 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio
 if asset_triggered_dags:
 self._create_dag_runs_asset_triggered(
 dag_models=[d for d in asset_triggered_dags if d.dag_id not in partition_dag_ids],
 triggered_date_by_dag=triggered_date_by_dag,
 session=session,
 )

Expand DownExpand Up@@ -2005,30 +2004,44 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -

 def _create_dag_runs_asset_triggered(
 self,
 *,
 dag_models: Collection[DagModel],
 triggered_date_by_dag: dict[str, datetime],
 session: Session,
 ) -> None:
 """For DAGs that are triggered by assets, create dag runs."""
 triggered_dates: dict[str, DateTime] = {
 dag_id: timezone.coerce_datetime(last_asset_event_time)
 for dag_id, last_asset_event_time in triggered_date_by_dag.items()
 }

 """For Dags that are triggered by assets, create Dag runs."""
 for dag_model in dag_models:
 dag = self._get_current_dag(dag_id=dag_model.dag_id, session=session)
 if not dag:
 self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
 self.log.error("Dag '%s' not found in serialized_dag table", dag_model.dag_id)
 continue

 if not isinstance(dag.timetable, AssetTriggeredTimetable):
 self.log.error(
 "DAG '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 "Dag '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 dag_model.dag_id,
 )
 continue

 triggered_date = triggered_dates[dag.dag_id]
 queued_adrqs = session.scalars(
 with_row_locks(
 select(AssetDagRunQueue)
 .where(AssetDagRunQueue.target_dag_id == dag.dag_id)
 .order_by(AssetDagRunQueue.created_at.desc()),
 of=AssetDagRunQueue,
 skip_locked=True,
 key_share=False,
 session=session,
 )
 ).all()
 # If another scheduler already locked these ADRQ rows, SKIP LOCKED makes this scheduler skip them.
 if not queued_adrqs:
 self.log.debug(
 "Skipping asset-triggered DagRun creation for Dag '%s'; no queued assets remain.",
 dag.dag_id,
 )
 continue

 triggered_date: DateTime = timezone.coerce_datetime(queued_adrqs[0].created_at)
 cte = (
 select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
 .where(
Expand DownExpand Up@@ -2077,7 +2090,15 @@ def _create_dag_runs_asset_triggered(
 )
 Stats.incr("asset.triggered_dagruns")
 dag_run.consumed_asset_events.extend(asset_events)
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))

 # 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).
 adrq_pks = [(record.asset_id, record.target_dag_id) for record in queued_adrqs]
 session.execute(
 delete(AssetDagRunQueue).where(
 tuple_(AssetDagRunQueue.asset_id, AssetDagRunQueue.target_dag_id).in_(adrq_pks)
 )
 )

 def _lock_backfills(self, dag_runs: Collection[DagRun], session: Session) -> dict[int, Backfill]:
 """
Expand Down

89 changes: 87 additions & 2 deletions
airflow-core/tests/unit/jobs/test_scheduler_job.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -285,9 +285,9 @@ def per_test(self) -> Generator:
 def set_instance_attrs(self) -> Generator:
 # Speed up some tests by not running the tasks, just look at what we
 # enqueue!
 self.null_exec: MockExecutor | None = MockExecutor()
 self.null_exec: BaseExecutor = MockExecutor()
 yield
 self.null_exec = None
 self.null_exec = None # type: ignore[assignment]

 @pytest.fixture
 def mock_executors(self):
Expand DownExpand Up@@ -4858,6 +4858,91 @@ def dict_from_obj(obj):

 assert created_run.creating_job_id == scheduler_job.id

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_skips_stale_triggered_date(self, session, dag_maker):
 asset = Asset(uri="test://asset-for-stale-trigger-date", name="asset-for-stale-trigger-date")
 with dag_maker(dag_id="asset-consumer-stale-trigger-date", schedule=[asset], session=session):
 pass
 dag_model = dag_maker.dag_model
 asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri == asset.uri))

 queued_at = timezone.utcnow()
 session.add(AssetDagRunQueue(target_dag_id=dag_model.dag_id, asset_id=asset_id, created_at=queued_at))
 session.flush()

 # Simulate another scheduler consuming ADRQ rows after we computed triggered_date_by_dag.
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_model.dag_id))
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 session=session,
 )

 # We do not create a new DagRun seems the ADRQ has already been consumed
 assert session.scalars(select(DagRun).where(DagRun.dag_id == dag_model.dag_id)).one_or_none() is None

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows(
 self, session: Session, dag_maker
 ):
 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):
 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))

 session.add_all(
 [
 # The ADRQ that should triggers the Dag run creation
 AssetDagRunQueue(
 asset_id=asset_1_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 # The ADRQ that arrives after the Dag run creation but before ADRQ clean up
 # This situation is simluarted by _lock_only_selected_asset below
 AssetDagRunQueue(
 asset_id=asset_2_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 ]
 )
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])

 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)

 with patch("airflow.jobs.scheduler_job_runner.with_row_locks", side_effect=_lock_only_selected_asset):
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 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

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
 """
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Remove or un-stick sticky/fixed headers that block content\n(function() {\n function unstick() {\n document.querySelectorAll('header, nav, [role=\"banner\"], .header, .navbar, .sticky, .fixed-top, [style*=\"position: fixed\"], [style*=\"position:sticky\"]').forEach(function(el) {\n if (el.style.position === 'fixed' || el.style.position === 'sticky' || \n getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {\n el.style.position = 'static';\n el.style.top = 'auto';\n el.style.zIndex = 'auto';\n }\n });\n }\n \n unstick();\n \n var observer = new MutationObserver(unstick);\n observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });\n})();", "Kill Sticky Headers");
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '

Skip to content
Add row lock to ADRQ before Dag run creation#60773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 6 commits into

from

Mar 17, 2026
 Merged


Add row lock to ADRQ before Dag run creation
#60773
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


6 commits


Select commit
Hold shift + click to select a range
659ea0c

fix(ADRQ): lock ADRQs right before Dag run creationg to avoid race co…


Lee-W Mar 2, 2026
56ec448

style: ignore MockExecutor as it intends to be a mocked BaseExecutor


Lee-W Mar 2, 2026
6c474d1

refactor: address review comments


Lee-W Mar 5, 2026
b866aeb

refactor: sort by created_by so that we don't need to use max again


Lee-W Mar 5, 2026
e140c41

refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026
51da4e1

fixup! refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


45 changes: 33 additions & 12 deletions
airflow-core/src/airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1838,7 +1838,6 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio
 if asset_triggered_dags:
 self._create_dag_runs_asset_triggered(
 dag_models=[d for d in asset_triggered_dags if d.dag_id not in partition_dag_ids],
 triggered_date_by_dag=triggered_date_by_dag,
 session=session,
 )

Expand DownExpand Up@@ -2005,30 +2004,44 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -

 def _create_dag_runs_asset_triggered(
 self,
 *,
 dag_models: Collection[DagModel],
 triggered_date_by_dag: dict[str, datetime],
 session: Session,
 ) -> None:
 """For DAGs that are triggered by assets, create dag runs."""
 triggered_dates: dict[str, DateTime] = {
 dag_id: timezone.coerce_datetime(last_asset_event_time)
 for dag_id, last_asset_event_time in triggered_date_by_dag.items()
 }

 """For Dags that are triggered by assets, create Dag runs."""
 for dag_model in dag_models:
 dag = self._get_current_dag(dag_id=dag_model.dag_id, session=session)
 if not dag:
 self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
 self.log.error("Dag '%s' not found in serialized_dag table", dag_model.dag_id)
 continue

 if not isinstance(dag.timetable, AssetTriggeredTimetable):
 self.log.error(
 "DAG '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 "Dag '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 dag_model.dag_id,
 )
 continue

 triggered_date = triggered_dates[dag.dag_id]
 queued_adrqs = session.scalars(
 with_row_locks(
 select(AssetDagRunQueue)
 .where(AssetDagRunQueue.target_dag_id == dag.dag_id)
 .order_by(AssetDagRunQueue.created_at.desc()),
 of=AssetDagRunQueue,
 skip_locked=True,
 key_share=False,
 session=session,
 )
 ).all()
 # If another scheduler already locked these ADRQ rows, SKIP LOCKED makes this scheduler skip them.
 if not queued_adrqs:
 self.log.debug(
 "Skipping asset-triggered DagRun creation for Dag '%s'; no queued assets remain.",
 dag.dag_id,
 )
 continue

 triggered_date: DateTime = timezone.coerce_datetime(queued_adrqs[0].created_at)
 cte = (
 select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
 .where(
Expand DownExpand Up@@ -2077,7 +2090,15 @@ def _create_dag_runs_asset_triggered(
 )
 Stats.incr("asset.triggered_dagruns")
 dag_run.consumed_asset_events.extend(asset_events)
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))

 # 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).
 adrq_pks = [(record.asset_id, record.target_dag_id) for record in queued_adrqs]
 session.execute(
 delete(AssetDagRunQueue).where(
 tuple_(AssetDagRunQueue.asset_id, AssetDagRunQueue.target_dag_id).in_(adrq_pks)
 )
 )

 def _lock_backfills(self, dag_runs: Collection[DagRun], session: Session) -> dict[int, Backfill]:
 """
Expand Down

89 changes: 87 additions & 2 deletions
airflow-core/tests/unit/jobs/test_scheduler_job.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -285,9 +285,9 @@ def per_test(self) -> Generator:
 def set_instance_attrs(self) -> Generator:
 # Speed up some tests by not running the tasks, just look at what we
 # enqueue!
 self.null_exec: MockExecutor | None = MockExecutor()
 self.null_exec: BaseExecutor = MockExecutor()
 yield
 self.null_exec = None
 self.null_exec = None # type: ignore[assignment]

 @pytest.fixture
 def mock_executors(self):
Expand DownExpand Up@@ -4858,6 +4858,91 @@ def dict_from_obj(obj):

 assert created_run.creating_job_id == scheduler_job.id

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_skips_stale_triggered_date(self, session, dag_maker):
 asset = Asset(uri="test://asset-for-stale-trigger-date", name="asset-for-stale-trigger-date")
 with dag_maker(dag_id="asset-consumer-stale-trigger-date", schedule=[asset], session=session):
 pass
 dag_model = dag_maker.dag_model
 asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri == asset.uri))

 queued_at = timezone.utcnow()
 session.add(AssetDagRunQueue(target_dag_id=dag_model.dag_id, asset_id=asset_id, created_at=queued_at))
 session.flush()

 # Simulate another scheduler consuming ADRQ rows after we computed triggered_date_by_dag.
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_model.dag_id))
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 session=session,
 )

 # We do not create a new DagRun seems the ADRQ has already been consumed
 assert session.scalars(select(DagRun).where(DagRun.dag_id == dag_model.dag_id)).one_or_none() is None

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows(
 self, session: Session, dag_maker
 ):
 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):
 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))

 session.add_all(
 [
 # The ADRQ that should triggers the Dag run creation
 AssetDagRunQueue(
 asset_id=asset_1_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 # The ADRQ that arrives after the Dag run creation but before ADRQ clean up
 # This situation is simluarted by _lock_only_selected_asset below
 AssetDagRunQueue(
 asset_id=asset_2_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 ]
 )
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])

 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)

 with patch("airflow.jobs.scheduler_job_runner.with_row_locks", side_effect=_lock_only_selected_asset):
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 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

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
 """
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
injectUserscript("// Universal Dark Mode - works on any site\n(function() {\n var enabled = true;\n \n function applyDarkMode() {\n if (!enabled) return;\n \n // Create style element if it doesn't exist\n var style = document.getElementById('universal-dark-mode-style');\n if (!style) {\n style = document.createElement('style');\n style.id = 'universal-dark-mode-style';\n document.head.appendChild(style);\n }\n \n // Dark mode CSS - inverts colors but preserves images/video\n style.textContent = '\n /* Invert everything except media */\n html {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #1a1a2e !important;\n }\n \n /* Restore images, videos, iframes, canvas */\n img, video, iframe, canvas, svg, picture, [style*=\"background-image\"] {\n filter: invert(1) hue-rotate(180deg) !important;\n }\n \n /* Preserve specific elements that should not be inverted */\n .no-dark-mode, .no-dark-mode *,\n [data-theme=\"light\"], [data-theme=\"light\"],\n .ace_editor, .ace_editor *,\n .CodeMirror, .CodeMirror *,\n .monaco-editor, .monaco-editor *,\n .markdown-body pre, .markdown-body pre *,\n .highlight, .highlight *,\n pre code, pre code * {\n filter: none !important;\n }\n \n /* Fix common UI elements */\n .modal, .popup, .dropdown-menu, .tooltip, .popover {\n filter: invert(1) hue-rotate(180deg) !important;\n background: #2d2d44 !important;\n border-color: #444 !important;\n }\n \n /* Scrollbars */\n ::-webkit-scrollbar { background: #1a1a2e !important; }\n ::-webkit-scrollbar-thumb { background: #444 !important; }\n ::-webkit-scrollbar-thumb:hover { background: #555 !important; }\n \n /* Selection */\n ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }\n ';\n }\n \n function removeDarkMode() {\n var style = document.getElementById('universal-dark-mode-style');\n if (style) style.remove();\n }\n \n // Toggle with Alt+Shift+D\n document.addEventListener('keydown', function(e) {\n if (e.altKey && e.shiftKey && e.key === 'D') {\n e.preventDefault();\n enabled = !enabled;\n if (enabled) {\n applyDarkMode();\n console.log('[Universal Dark Mode] Enabled');\n } else {\n removeDarkMode();\n console.log('[Universal Dark Mode] Disabled');\n }\n }\n });\n \n // Apply on load\n applyDarkMode();\n \n // Re-apply on dynamic content\n var observer = new MutationObserver(function(mutations) {\n if (enabled && !document.getElementById('universal-dark-mode-style')) {\n applyDarkMode();\n }\n });\n observer.observe(document.head, { childList: true });\n \n console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');\n})();", "Universal Dark Mode");
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();

Skip to content
Add row lock to ADRQ before Dag run creation#60773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Lee-W
merged 6 commits into

from

Mar 17, 2026
 Merged


Add row lock to ADRQ before Dag run creation
#60773
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


6 commits


Select commit
Hold shift + click to select a range
659ea0c

fix(ADRQ): lock ADRQs right before Dag run creationg to avoid race co…


Lee-W Mar 2, 2026
56ec448

style: ignore MockExecutor as it intends to be a mocked BaseExecutor


Lee-W Mar 2, 2026
6c474d1

refactor: address review comments


Lee-W Mar 5, 2026
b866aeb

refactor: sort by created_by so that we don't need to use max again


Lee-W Mar 5, 2026
e140c41

refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026
51da4e1

fixup! refactor: remove redundant triggered_date_by_dag usage


Lee-W Mar 16, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


45 changes: 33 additions & 12 deletions
airflow-core/src/airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -1838,7 +1838,6 @@ def _create_dagruns_for_dags(self, guard: CommitProhibitorGuard, session: Sessio
 if asset_triggered_dags:
 self._create_dag_runs_asset_triggered(
 dag_models=[d for d in asset_triggered_dags if d.dag_id not in partition_dag_ids],
 triggered_date_by_dag=triggered_date_by_dag,
 session=session,
 )

Expand DownExpand Up@@ -2005,30 +2004,44 @@ def _create_dag_runs(self, dag_models: Collection[DagModel], session: Session) -

 def _create_dag_runs_asset_triggered(
 self,
 *,
 dag_models: Collection[DagModel],
 triggered_date_by_dag: dict[str, datetime],
 session: Session,
 ) -> None:
 """For DAGs that are triggered by assets, create dag runs."""
 triggered_dates: dict[str, DateTime] = {
 dag_id: timezone.coerce_datetime(last_asset_event_time)
 for dag_id, last_asset_event_time in triggered_date_by_dag.items()
 }

 """For Dags that are triggered by assets, create Dag runs."""
 for dag_model in dag_models:
 dag = self._get_current_dag(dag_id=dag_model.dag_id, session=session)
 if not dag:
 self.log.error("DAG '%s' not found in serialized_dag table", dag_model.dag_id)
 self.log.error("Dag '%s' not found in serialized_dag table", dag_model.dag_id)
 continue

 if not isinstance(dag.timetable, AssetTriggeredTimetable):
 self.log.error(
 "DAG '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 "Dag '%s' was asset-scheduled, but didn't have an AssetTriggeredTimetable!",
 dag_model.dag_id,
 )
 continue

 triggered_date = triggered_dates[dag.dag_id]
 queued_adrqs = session.scalars(
 with_row_locks(
 select(AssetDagRunQueue)
 .where(AssetDagRunQueue.target_dag_id == dag.dag_id)
 .order_by(AssetDagRunQueue.created_at.desc()),
 of=AssetDagRunQueue,
 skip_locked=True,
 key_share=False,
 session=session,
 )
 ).all()
 # If another scheduler already locked these ADRQ rows, SKIP LOCKED makes this scheduler skip them.
 if not queued_adrqs:
 self.log.debug(
 "Skipping asset-triggered DagRun creation for Dag '%s'; no queued assets remain.",
 dag.dag_id,
 )
 continue

 triggered_date: DateTime = timezone.coerce_datetime(queued_adrqs[0].created_at)
 cte = (
 select(func.max(DagRun.run_after).label("previous_dag_run_run_after"))
 .where(
Expand DownExpand Up@@ -2077,7 +2090,15 @@ def _create_dag_runs_asset_triggered(
 )
 Stats.incr("asset.triggered_dagruns")
 dag_run.consumed_asset_events.extend(asset_events)
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_run.dag_id))

 # 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).
 adrq_pks = [(record.asset_id, record.target_dag_id) for record in queued_adrqs]
 session.execute(
 delete(AssetDagRunQueue).where(
 tuple_(AssetDagRunQueue.asset_id, AssetDagRunQueue.target_dag_id).in_(adrq_pks)
 )
 )

 def _lock_backfills(self, dag_runs: Collection[DagRun], session: Session) -> dict[int, Backfill]:
 """
Expand Down

89 changes: 87 additions & 2 deletions
airflow-core/tests/unit/jobs/test_scheduler_job.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -285,9 +285,9 @@ def per_test(self) -> Generator:
 def set_instance_attrs(self) -> Generator:
 # Speed up some tests by not running the tasks, just look at what we
 # enqueue!
 self.null_exec: MockExecutor | None = MockExecutor()
 self.null_exec: BaseExecutor = MockExecutor()
 yield
 self.null_exec = None
 self.null_exec = None # type: ignore[assignment]

 @pytest.fixture
 def mock_executors(self):
Expand DownExpand Up@@ -4858,6 +4858,91 @@ def dict_from_obj(obj):

 assert created_run.creating_job_id == scheduler_job.id

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_skips_stale_triggered_date(self, session, dag_maker):
 asset = Asset(uri="test://asset-for-stale-trigger-date", name="asset-for-stale-trigger-date")
 with dag_maker(dag_id="asset-consumer-stale-trigger-date", schedule=[asset], session=session):
 pass
 dag_model = dag_maker.dag_model
 asset_id = session.scalar(select(AssetModel.id).where(AssetModel.uri == asset.uri))

 queued_at = timezone.utcnow()
 session.add(AssetDagRunQueue(target_dag_id=dag_model.dag_id, asset_id=asset_id, created_at=queued_at))
 session.flush()

 # Simulate another scheduler consuming ADRQ rows after we computed triggered_date_by_dag.
 session.execute(delete(AssetDagRunQueue).where(AssetDagRunQueue.target_dag_id == dag_model.dag_id))
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 session=session,
 )

 # We do not create a new DagRun seems the ADRQ has already been consumed
 assert session.scalars(select(DagRun).where(DagRun.dag_id == dag_model.dag_id)).one_or_none() is None

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_triggered_deletes_only_selected_adrq_rows(
 self, session: Session, dag_maker
 ):
 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):
 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))

 session.add_all(
 [
 # The ADRQ that should triggers the Dag run creation
 AssetDagRunQueue(
 asset_id=asset_1_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 # The ADRQ that arrives after the Dag run creation but before ADRQ clean up
 # This situation is simluarted by _lock_only_selected_asset below
 AssetDagRunQueue(
 asset_id=asset_2_id, target_dag_id=dag_model.dag_id, created_at=timezone.utcnow()
 ),
 ]
 )
 session.flush()

 scheduler_job = Job()
 self.job_runner = SchedulerJobRunner(job=scheduler_job, executors=[self.null_exec])

 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)

 with patch("airflow.jobs.scheduler_job_runner.with_row_locks", side_effect=_lock_only_selected_asset):
 self.job_runner._create_dag_runs_asset_triggered(
 dag_models=[dag_model],
 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

 @pytest.mark.need_serialized_dag
 def test_create_dag_runs_asset_alias_with_asset_event_attached(self, session, dag_maker):
 """
Expand Down
Loading