You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "fat-catTW (via GitHub)" <gi...@apache.org> on 2026/08/06 06:48:35 UTC
[PR] Store executor callback DagRun references in the database [airflow]
fat-catTW opened a new pull request, #71214:
URL: https://github.com/apache/airflow/pull/71214
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: https://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
## Why
Executor callbacks need to know which DagRun they belong to when the scheduler
builds callback workloads. Keeping that reference only in `callback.data` makes
the scheduler depend on payload shape instead of a database-managed relationship.
## Solution
Add a nullable `callback.dagrun_id` foreign key to `dag_run.id` and have executor
callbacks store their DagRun reference there. The scheduler now loads the
relationship through the ORM when enqueueing callback workloads.
The migration backfills existing rows from deadline relationships first, then
falls back to legacy callback payload data when needed.
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- `[X]` Yes (please specify the tool below)
Generated-by: [Codex] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "fat-catTW (via GitHub)" <gi...@apache.org>.
fat-catTW commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r4337108849
##########
airflow-core/src/airflow/migrations/versions/0149_3_5_0_add_dagrun_id_to_callback.py:
##########
@@ -0,0 +1,167 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add dagrun_id foreign key to callback.
+
+Revision ID: 0f4c9a2b8e1d
+Revises: c71065a1b14f
+Create Date: 2026-08-06 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import json
+
+import sqlalchemy as sa
+from alembic import context, op
+
+from airflow.configuration import conf
+
+# revision identifiers, used by Alembic.
+revision = "0f4c9a2b8e1d"
+down_revision = "c71065a1b14f"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+
+def _deserialize_data(data):
+ if isinstance(data, str):
+ try:
+ data = json.loads(data)
+ except json.JSONDecodeError:
+ return {}
+ return data.get("__var", data) if isinstance(data, dict) else {}
+
+
+def _extract_dagrun_id(data) -> int | None:
+ dagrun_id = _deserialize_data(data).get("dag_run_id")
+ if dagrun_id is None:
+ return None
+ try:
+ return int(dagrun_id)
+ except (TypeError, ValueError):
+ return None
+
+
+def _backfill_dagrun_id_from_deadline(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ deadline = sa.table(
+ "deadline",
+ sa.column("callback_id", sa.Uuid()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, deadline.c.dagrun_id)
+ .select_from(
+ callback.join(deadline, callback.c.id == deadline.c.callback_id).join(
+ dag_run, deadline.c.dagrun_id == dag_run.c.id
+ )
+ )
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, dagrun_id in rows:
+ conn.execute(callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id))
+
+ last_id = rows[-1].id
+
+
+def _backfill_dagrun_id(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("data", sa.Text()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, callback.c.data)
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, data in rows:
+ dagrun_id = _extract_dagrun_id(data)
+ if dagrun_id is None:
+ continue
+
+ dagrun_exists = conn.execute(
+ sa.select(dag_run.c.id).where(dag_run.c.id == dagrun_id).limit(1)
+ ).first()
+ if dagrun_exists:
+ conn.execute(
+ callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id)
+ )
+
+ last_id = rows[-1].id
+
+
+def upgrade():
+ """Add callback.dagrun_id and backfill it from legacy callback.data."""
+ with op.batch_alter_table("callback") as batch_op:
+ batch_op.add_column(sa.Column("dagrun_id", sa.Integer(), nullable=True))
+
+ if not context.is_offline_mode():
+ conn = op.get_bind()
+ batch_size = conf.getint("database", "migration_batch_size")
+ _backfill_dagrun_id_from_deadline(conn, batch_size)
+ _backfill_dagrun_id(conn, batch_size)
+
+ with op.batch_alter_table("callback") as batch_op:
+ batch_op.create_index("callback_dagrun_id_idx", ["dagrun_id"], unique=False)
+ batch_op.create_foreign_key(
Review Comment:
This was a real SQLite data-loss risk. Thanks for pointing it out.
I wrapped the full upgrade and downgrade bodies in disable_sqlite_fkeys(op), before any DML or batch table operation. I also added a SQLite upgrade/downgrade test with deadline.callback_id ON DELETE CASCADE enabled, which verifies that the deadline row is preserved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r3744887964
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1268,24 +1269,11 @@ def _enqueue_executor_callbacks(self, session: Session) -> None:
# Can't happen since we queried ExecutorCallback, but satisfies mypy.
continue
- # TODO: Add dagrun_id as a proper ORM foreign key on the callback table instead of storing in data dict.
- # This would eliminate this reconstruction step. For now, all ExecutorCallbacks
- # are expected to have dag_run_id set in their data dict (e.g., by Deadline.handle_miss).
- if not isinstance(callback.data, dict) or "dag_run_id" not in callback.data:
- self.log.error(
- "ExecutorCallback %s is missing required 'dag_run_id' in data dict. "
- "This indicates a bug in callback creation. Skipping callback.",
- callback.id,
- )
- continue
-
- dag_run_id = callback.data["dag_run_id"]
- dag_run = session.get(DagRun, dag_run_id)
-
+ dag_run = callback.dag_run
if dag_run is None:
self.log.warning(
"Could not find DagRun with id=%s for callback %s. DagRun may have been deleted.",
- dag_run_id,
+ callback.dagrun_id,
callback.id,
)
Review Comment:
This collapses two previously distinct failure modes (missing `dagrun_id` due to a callback-creation bug vs. a DagRun deleted after enqueue) into one warning path. Only one production call site sets `dagrun_id` today, so it's not reachable now, but it'd be easy for a future ExecutorCallback creation path to forget to set it and silently emits "DagRun may have been deleted."
Maybe add different checks to `callback.dagrun_id is None` and `dag_run is None` with dagrun_id set?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "fat-catTW (via GitHub)" <gi...@apache.org>.
fat-catTW commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r4337084753
##########
airflow-core/src/airflow/migrations/versions/0149_3_5_0_add_dagrun_id_to_callback.py:
##########
@@ -0,0 +1,167 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add dagrun_id foreign key to callback.
+
+Revision ID: 0f4c9a2b8e1d
+Revises: c71065a1b14f
+Create Date: 2026-08-06 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import json
+
+import sqlalchemy as sa
+from alembic import context, op
+
+from airflow.configuration import conf
+
+# revision identifiers, used by Alembic.
+revision = "0f4c9a2b8e1d"
+down_revision = "c71065a1b14f"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+
+def _deserialize_data(data):
+ if isinstance(data, str):
+ try:
+ data = json.loads(data)
+ except json.JSONDecodeError:
+ return {}
+ return data.get("__var", data) if isinstance(data, dict) else {}
+
+
+def _extract_dagrun_id(data) -> int | None:
+ dagrun_id = _deserialize_data(data).get("dag_run_id")
+ if dagrun_id is None:
+ return None
+ try:
+ return int(dagrun_id)
+ except (TypeError, ValueError):
+ return None
+
+
+def _backfill_dagrun_id_from_deadline(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ deadline = sa.table(
+ "deadline",
+ sa.column("callback_id", sa.Uuid()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, deadline.c.dagrun_id)
+ .select_from(
+ callback.join(deadline, callback.c.id == deadline.c.callback_id).join(
+ dag_run, deadline.c.dagrun_id == dag_run.c.id
+ )
+ )
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, dagrun_id in rows:
+ conn.execute(callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id))
+
+ last_id = rows[-1].id
+
+
+def _backfill_dagrun_id(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("data", sa.Text()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, callback.c.data)
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, data in rows:
+ dagrun_id = _extract_dagrun_id(data)
+ if dagrun_id is None:
+ continue
+
+ dagrun_exists = conn.execute(
+ sa.select(dag_run.c.id).where(dag_run.c.id == dagrun_id).limit(1)
+ ).first()
+ if dagrun_exists:
+ conn.execute(
+ callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id)
+ )
+
+ last_id = rows[-1].id
+
+
+def upgrade():
+ """Add callback.dagrun_id and backfill it from legacy callback.data."""
+ with op.batch_alter_table("callback") as batch_op:
+ batch_op.add_column(sa.Column("dagrun_id", sa.Integer(), nullable=True))
+
+ if not context.is_offline_mode():
+ conn = op.get_bind()
+ batch_size = conf.getint("database", "migration_batch_size")
+ _backfill_dagrun_id_from_deadline(conn, batch_size)
+ _backfill_dagrun_id(conn, batch_size)
+
+ with op.batch_alter_table("callback") as batch_op:
+ batch_op.create_index("callback_dagrun_id_idx", ["dagrun_id"], unique=False)
+ batch_op.create_foreign_key(
+ batch_op.f("callback_dagrun_id_fkey"), "dag_run", ["dagrun_id"], ["id"], ondelete="CASCADE"
+ )
+
+
+def downgrade():
Review Comment:
Thanks for review!
Before dropping the column, the downgrade now restores dagrun_id into the legacy callback data. I also added a test covering both wrapped and plain callback data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "fat-catTW (via GitHub)" <gi...@apache.org>.
fat-catTW commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r4337071072
##########
airflow-core/src/airflow/models/callback.py:
##########
@@ -143,10 +143,17 @@ class Callback(Base, BaseWorkload):
# Creation time of the callback
created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=timezone.utcnow, nullable=False)
+ dagrun_id: Mapped[int | None] = mapped_column(
+ Integer, ForeignKey("dag_run.id", ondelete="CASCADE"), nullable=True
Review Comment:
Thanks, good point.
I added callback to dag_run's dependent tables so db clean -t dag_run archives callbacks before deleting the DagRun. I also added an integration test that creates an executor callback linked to a DagRun and verifies that it is archived.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "fat-catTW (via GitHub)" <gi...@apache.org>.
fat-catTW commented on PR #71214:
URL: https://github.com/apache/airflow/pull/71214#issuecomment-6209398618
Thanks for checking the PR.
I have rebased the PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #71214:
URL: https://github.com/apache/airflow/pull/71214#issuecomment-6197991652
I've merged everything in the chain that was blocking this; if you can get the migration rebased I can approve it and get it merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r4324794803
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1366,25 +1367,19 @@ def _enqueue_executor_callbacks(self, session: Session) -> None:
# Can't happen since we queried ExecutorCallback, but satisfies mypy.
continue
- # TODO: Add dagrun_id as a proper ORM foreign key on the callback table instead of storing in data dict.
- # This would eliminate this reconstruction step. For now, all ExecutorCallbacks
- # are expected to have dag_run_id set in their data dict (e.g., by Deadline.handle_miss).
- if not isinstance(callback.data, dict) or "dag_run_id" not in callback.data:
- self.log.error(
- "ExecutorCallback %s is missing required 'dag_run_id' in data dict. "
- "This indicates a bug in callback creation. Skipping callback.",
- callback.id,
+ if callback.dagrun_id is None:
+ self.log.warning(
Review Comment:
Maybe not related to this PR (cause I think it already exists today) but I am wondering in which case we end up in this situation.
1. Is that an expected scenario? If no, why a warning?
2. If yes, how come we ended up there?
I am asking because it feels weird to me to log a warning (that nobody will read) for something that feels wrong?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r4367504518
##########
airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dagrun_id_to_callback.py:
##########
@@ -0,0 +1,219 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add dagrun_id foreign key to callback.
+
+Revision ID: 0f4c9a2b8e1d
+Revises: c71065a1b14f
+Create Date: 2026-08-06 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import json
+
+import sqlalchemy as sa
+from alembic import context, op
+
+from airflow.configuration import conf
+from airflow.migrations.utils import disable_sqlite_fkeys
+
+# revision identifiers, used by Alembic.
+revision = "0f4c9a2b8e1d"
+down_revision = "c71065a1b14f"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+
+def _deserialize_data(data):
+ if isinstance(data, str):
+ try:
+ data = json.loads(data)
+ except json.JSONDecodeError:
+ return {}
+ return data.get("__var", data) if isinstance(data, dict) else {}
+
+
+def _extract_dagrun_id(data) -> int | None:
+ dagrun_id = _deserialize_data(data).get("dag_run_id")
+ if dagrun_id is None:
+ return None
+ try:
+ return int(dagrun_id)
+ except (TypeError, ValueError):
+ return None
+
+
+def _backfill_dagrun_id_from_deadline(conn, batch_size: int) -> None:
Review Comment:
I'm not sure if this a good idea, open to discuss.
Would it be better to _not_ backfill the `callback.dagrun_id` at all?
Since we could add the fallback at the read side (`scheduler_job_runner.py`) and note it down properly that we shouldn't remove the fallback on `callback.data`.
##########
airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dagrun_id_to_callback.py:
##########
@@ -0,0 +1,219 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add dagrun_id foreign key to callback.
+
+Revision ID: 0f4c9a2b8e1d
+Revises: c71065a1b14f
+Create Date: 2026-08-06 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import json
+
+import sqlalchemy as sa
+from alembic import context, op
+
+from airflow.configuration import conf
+from airflow.migrations.utils import disable_sqlite_fkeys
+
+# revision identifiers, used by Alembic.
+revision = "0f4c9a2b8e1d"
+down_revision = "c71065a1b14f"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+
+def _deserialize_data(data):
+ if isinstance(data, str):
+ try:
+ data = json.loads(data)
+ except json.JSONDecodeError:
+ return {}
+ return data.get("__var", data) if isinstance(data, dict) else {}
+
+
+def _extract_dagrun_id(data) -> int | None:
+ dagrun_id = _deserialize_data(data).get("dag_run_id")
+ if dagrun_id is None:
+ return None
+ try:
+ return int(dagrun_id)
+ except (TypeError, ValueError):
+ return None
+
+
+def _backfill_dagrun_id_from_deadline(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ deadline = sa.table(
+ "deadline",
+ sa.column("callback_id", sa.Uuid()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, deadline.c.dagrun_id)
+ .select_from(
+ callback.join(deadline, callback.c.id == deadline.c.callback_id).join(
+ dag_run, deadline.c.dagrun_id == dag_run.c.id
+ )
+ )
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, dagrun_id in rows:
+ conn.execute(callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id))
+
+ last_id = rows[-1].id
+
+
+def _backfill_dagrun_id(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("data", sa.Text()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, callback.c.data)
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, data in rows:
+ dagrun_id = _extract_dagrun_id(data)
+ if dagrun_id is None:
+ continue
+
+ dagrun_exists = conn.execute(
+ sa.select(dag_run.c.id).where(dag_run.c.id == dagrun_id).limit(1)
+ ).first()
+ if dagrun_exists:
+ conn.execute(
+ callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id)
+ )
+
+ last_id = rows[-1].id
+
+
+def _restore_legacy_dagrun_id(conn, batch_size: int) -> None:
Review Comment:
Please noted that event we don't backfill for the upgrade path, we still need to keep the downgrade path backfill.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "fat-catTW (via GitHub)" <gi...@apache.org>.
fat-catTW commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r3747182696
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1268,24 +1269,11 @@ def _enqueue_executor_callbacks(self, session: Session) -> None:
# Can't happen since we queried ExecutorCallback, but satisfies mypy.
continue
- # TODO: Add dagrun_id as a proper ORM foreign key on the callback table instead of storing in data dict.
- # This would eliminate this reconstruction step. For now, all ExecutorCallbacks
- # are expected to have dag_run_id set in their data dict (e.g., by Deadline.handle_miss).
- if not isinstance(callback.data, dict) or "dag_run_id" not in callback.data:
- self.log.error(
- "ExecutorCallback %s is missing required 'dag_run_id' in data dict. "
- "This indicates a bug in callback creation. Skipping callback.",
- callback.id,
- )
- continue
-
- dag_run_id = callback.data["dag_run_id"]
- dag_run = session.get(DagRun, dag_run_id)
-
+ dag_run = callback.dag_run
if dag_run is None:
self.log.warning(
"Could not find DagRun with id=%s for callback %s. DagRun may have been deleted.",
- dag_run_id,
+ callback.dagrun_id,
callback.id,
)
Review Comment:
Good point, thank you for reviewing.
I split the checks so a missing `callback.dagrun_id` is reported separately from the case where `dagrun_id` is set but the related DagRun cannot be loaded.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "fat-catTW (via GitHub)" <gi...@apache.org>.
fat-catTW commented on PR #71214:
URL: https://github.com/apache/airflow/pull/71214#issuecomment-6397898881
I benchmarked the current backfill approach, and I would prefer to keep it without a read-side fallback to `callback.data`.
Environment:
- Breeze
- PostgreSQL 14
- Synthetic legacy executor callbacks
- `migration_batch_size=1,000`
- Each scenario was run once
| Rows | Legacy-data distribution | Upgrade: deadline | Upgrade: payload | Upgrade: total | Downgrade |
| ---: | --- | ---: | ---: | ---: | ---: |
| 10,000 | 100% deadline | 2.91s | 0.00s | 2.91s | 3.36s |
| 10,000 | 50% deadline / 50% payload | 0.97s | 2.05s | 3.02s | 2.13s |
| 10,000 | 100% payload | 0.01s | 3.68s | 3.69s | 1.80s |
| 100,000 | 100% deadline | 38.18s | 0.01s | 38.19s | 16.61s |
| 100,000 | 50% deadline / 50% payload | 13.95s | 33.55s | 47.50s | 16.59s |
| 100,000 | 100% payload | 0.04s | 55.73s | 55.77s | 17.14s |
| 1,000,000 | 100% deadline | 3m 03s | 0.32s | 3m 03s | 2m 58s |
| 1,000,000 | 50% deadline / 50% payload | 1m 33s | 2m 58s | 4m 31s | 2m 55s |
| 1,000,000 | 100% payload | 0.00s | 6m 05s | 6m 05s | 2m 58s |
Note: During upgrade, the migration first backfills from `deadline.dagrun_id`, then falls back to legacy `callback.data["dag_run_id"]` for remaining callbacks.
All scenarios backfilled every callback successfully.
This is a local synthetic benchmark, so production hardware, concurrent workload, replication, and real data distribution may differ. However, it suggests the payload-only worst case is a bounded, one-time upgrade cost of about six minutes at one million rows.
Based on this test, I think keeping the one-time backfill and avoiding a permanent read-side fallback is a reasonable trade-off. It lets the Scheduler rely consistently on the database-managed `callback.dagrun_id` relationship instead of continuing to support the legacy payload representation.
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "fat-catTW (via GitHub)" <gi...@apache.org>.
fat-catTW commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r4328145995
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1366,25 +1367,19 @@ def _enqueue_executor_callbacks(self, session: Session) -> None:
# Can't happen since we queried ExecutorCallback, but satisfies mypy.
continue
- # TODO: Add dagrun_id as a proper ORM foreign key on the callback table instead of storing in data dict.
- # This would eliminate this reconstruction step. For now, all ExecutorCallbacks
- # are expected to have dag_run_id set in their data dict (e.g., by Deadline.handle_miss).
- if not isinstance(callback.data, dict) or "dag_run_id" not in callback.data:
- self.log.error(
- "ExecutorCallback %s is missing required 'dag_run_id' in data dict. "
- "This indicates a bug in callback creation. Skipping callback.",
- callback.id,
+ if callback.dagrun_id is None:
+ self.log.warning(
Review Comment:
Thanks for the review!
This is not an expected state for an executor callback. The deadline path always assigns its DagRun reference.
It can still occur for a legacy row that cannot be backfilled, or if a future callback producer omits the reference. I changed the log from warning to error and added a test that asserts the error-level log.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "ramitkataria (via GitHub)" <gi...@apache.org>.
ramitkataria commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r4331611151
##########
airflow-core/src/airflow/models/callback.py:
##########
@@ -143,10 +143,17 @@ class Callback(Base, BaseWorkload):
# Creation time of the callback
created_at: Mapped[datetime] = mapped_column(UtcDateTime, default=timezone.utcnow, nullable=False)
+ dagrun_id: Mapped[int | None] = mapped_column(
+ Integer, ForeignKey("dag_run.id", ondelete="CASCADE"), nullable=True
Review Comment:
`db_cleanup.py` isn't in the diff so commenting here: Should `callback` be added to `dag_run`'s `dependent_tables` in `db_cleanup.py`, like `task_state_store` was in #68218 and `task_reschedule` in #71990? With this CASCADE, I think `db clean -t dag_run` would delete the run's callbacks without archiving them
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1366,25 +1367,19 @@ def _enqueue_executor_callbacks(self, session: Session) -> None:
# Can't happen since we queried ExecutorCallback, but satisfies mypy.
continue
- # TODO: Add dagrun_id as a proper ORM foreign key on the callback table instead of storing in data dict.
- # This would eliminate this reconstruction step. For now, all ExecutorCallbacks
- # are expected to have dag_run_id set in their data dict (e.g., by Deadline.handle_miss).
- if not isinstance(callback.data, dict) or "dag_run_id" not in callback.data:
- self.log.error(
- "ExecutorCallback %s is missing required 'dag_run_id' in data dict. "
- "This indicates a bug in callback creation. Skipping callback.",
- callback.id,
+ if callback.dagrun_id is None:
+ self.log.warning(
Review Comment:
Thanks for switching to error. I don't think anything fills `dagrun_id` in later, so the callback stays PENDING and gets fetched and logged again every loop. With the query taking the top `max_callbacks` by `priority_weight`, enough of these could crowd out valid callbacks. Maybe we should mark it `FAILED` with an `output`, similar to how the scheduler fails task instances whose serialized Dag is missing?
Separately, maybe we should also have `Deadline.__init__` set `dagrun_id` next to `bundle_name` to avoid running into this?
##########
airflow-core/src/airflow/migrations/versions/0149_3_5_0_add_dagrun_id_to_callback.py:
##########
@@ -0,0 +1,167 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add dagrun_id foreign key to callback.
+
+Revision ID: 0f4c9a2b8e1d
+Revises: c71065a1b14f
+Create Date: 2026-08-06 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import json
+
+import sqlalchemy as sa
+from alembic import context, op
+
+from airflow.configuration import conf
+
+# revision identifiers, used by Alembic.
+revision = "0f4c9a2b8e1d"
+down_revision = "c71065a1b14f"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+
+def _deserialize_data(data):
+ if isinstance(data, str):
+ try:
+ data = json.loads(data)
+ except json.JSONDecodeError:
+ return {}
+ return data.get("__var", data) if isinstance(data, dict) else {}
+
+
+def _extract_dagrun_id(data) -> int | None:
+ dagrun_id = _deserialize_data(data).get("dag_run_id")
+ if dagrun_id is None:
+ return None
+ try:
+ return int(dagrun_id)
+ except (TypeError, ValueError):
+ return None
+
+
+def _backfill_dagrun_id_from_deadline(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ deadline = sa.table(
+ "deadline",
+ sa.column("callback_id", sa.Uuid()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, deadline.c.dagrun_id)
+ .select_from(
+ callback.join(deadline, callback.c.id == deadline.c.callback_id).join(
+ dag_run, deadline.c.dagrun_id == dag_run.c.id
+ )
+ )
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, dagrun_id in rows:
+ conn.execute(callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id))
+
+ last_id = rows[-1].id
+
+
+def _backfill_dagrun_id(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("data", sa.Text()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, callback.c.data)
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, data in rows:
+ dagrun_id = _extract_dagrun_id(data)
+ if dagrun_id is None:
+ continue
+
+ dagrun_exists = conn.execute(
+ sa.select(dag_run.c.id).where(dag_run.c.id == dagrun_id).limit(1)
+ ).first()
+ if dagrun_exists:
+ conn.execute(
+ callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id)
+ )
+
+ last_id = rows[-1].id
+
+
+def upgrade():
+ """Add callback.dagrun_id and backfill it from legacy callback.data."""
+ with op.batch_alter_table("callback") as batch_op:
+ batch_op.add_column(sa.Column("dagrun_id", sa.Integer(), nullable=True))
+
+ if not context.is_offline_mode():
+ conn = op.get_bind()
+ batch_size = conf.getint("database", "migration_batch_size")
+ _backfill_dagrun_id_from_deadline(conn, batch_size)
+ _backfill_dagrun_id(conn, batch_size)
+
+ with op.batch_alter_table("callback") as batch_op:
+ batch_op.create_index("callback_dagrun_id_idx", ["dagrun_id"], unique=False)
+ batch_op.create_foreign_key(
+ batch_op.f("callback_dagrun_id_fkey"), "dag_run", ["dagrun_id"], ["id"], ondelete="CASCADE"
+ )
+
+
+def downgrade():
Review Comment:
We might want to write `dagrun_id` back into `data["dag_run_id"]` before dropping the column. The scheduler code before this change reads that key and skips the callback without it, so executor callbacks still PENDING at downgrade time might never run
##########
airflow-core/src/airflow/migrations/versions/0149_3_5_0_add_dagrun_id_to_callback.py:
##########
@@ -0,0 +1,167 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add dagrun_id foreign key to callback.
+
+Revision ID: 0f4c9a2b8e1d
+Revises: c71065a1b14f
+Create Date: 2026-08-06 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import json
+
+import sqlalchemy as sa
+from alembic import context, op
+
+from airflow.configuration import conf
+
+# revision identifiers, used by Alembic.
+revision = "0f4c9a2b8e1d"
+down_revision = "c71065a1b14f"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+
+def _deserialize_data(data):
+ if isinstance(data, str):
+ try:
+ data = json.loads(data)
+ except json.JSONDecodeError:
+ return {}
+ return data.get("__var", data) if isinstance(data, dict) else {}
+
+
+def _extract_dagrun_id(data) -> int | None:
+ dagrun_id = _deserialize_data(data).get("dag_run_id")
+ if dagrun_id is None:
+ return None
+ try:
+ return int(dagrun_id)
+ except (TypeError, ValueError):
+ return None
+
+
+def _backfill_dagrun_id_from_deadline(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ deadline = sa.table(
+ "deadline",
+ sa.column("callback_id", sa.Uuid()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, deadline.c.dagrun_id)
+ .select_from(
+ callback.join(deadline, callback.c.id == deadline.c.callback_id).join(
+ dag_run, deadline.c.dagrun_id == dag_run.c.id
+ )
+ )
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, dagrun_id in rows:
+ conn.execute(callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id))
+
+ last_id = rows[-1].id
+
+
+def _backfill_dagrun_id(conn, batch_size: int) -> None:
+ callback = sa.table(
+ "callback",
+ sa.column("id", sa.Uuid()),
+ sa.column("type", sa.String(20)),
+ sa.column("data", sa.Text()),
+ sa.column("dagrun_id", sa.Integer()),
+ )
+ dag_run = sa.table("dag_run", sa.column("id", sa.Integer()))
+
+ last_id = None
+ while True:
+ query = (
+ sa.select(callback.c.id, callback.c.data)
+ .where(callback.c.type == "executor", callback.c.dagrun_id.is_(None))
+ .order_by(callback.c.id)
+ .limit(batch_size)
+ )
+ if last_id is not None:
+ query = query.where(callback.c.id > last_id)
+
+ rows = conn.execute(query).fetchall()
+ if not rows:
+ return
+
+ for callback_id, data in rows:
+ dagrun_id = _extract_dagrun_id(data)
+ if dagrun_id is None:
+ continue
+
+ dagrun_exists = conn.execute(
+ sa.select(dag_run.c.id).where(dag_run.c.id == dagrun_id).limit(1)
+ ).first()
+ if dagrun_exists:
+ conn.execute(
+ callback.update().where(callback.c.id == callback_id).values(dagrun_id=dagrun_id)
+ )
+
+ last_id = rows[-1].id
+
+
+def upgrade():
+ """Add callback.dagrun_id and backfill it from legacy callback.data."""
+ with op.batch_alter_table("callback") as batch_op:
+ batch_op.add_column(sa.Column("dagrun_id", sa.Integer(), nullable=True))
+
+ if not context.is_offline_mode():
+ conn = op.get_bind()
+ batch_size = conf.getint("database", "migration_batch_size")
+ _backfill_dagrun_id_from_deadline(conn, batch_size)
+ _backfill_dagrun_id(conn, batch_size)
+
+ with op.batch_alter_table("callback") as batch_op:
+ batch_op.create_index("callback_dagrun_id_idx", ["dagrun_id"], unique=False)
+ batch_op.create_foreign_key(
Review Comment:
On SQLite I think this empties the deadline table: adding the FK makes Alembic rebuild `callback`, and with `PRAGMA foreign_keys=ON` the `DROP TABLE` cascades through `deadline.callback_id`. In my local testing, `upgradedb` took `deadline` from 2 rows to 0, and `downgrade()` did the same. I think there's some guidelines on dealing with this in `contributing-docs/26_migration_round_trip_check.rst`. The round-trip CI job doesn't seed `callback`/`deadline`, so I don't think it would catch this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Store executor callback DagRun references in the database [airflow]
Posted by "fat-catTW (via GitHub)" <gi...@apache.org>.
fat-catTW commented on code in PR #71214:
URL: https://github.com/apache/airflow/pull/71214#discussion_r4370936143
##########
airflow-core/src/airflow/migrations/versions/0150_3_5_0_add_dagrun_id_to_callback.py:
##########
@@ -0,0 +1,219 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""
+Add dagrun_id foreign key to callback.
+
+Revision ID: 0f4c9a2b8e1d
+Revises: c71065a1b14f
+Create Date: 2026-08-06 00:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import json
+
+import sqlalchemy as sa
+from alembic import context, op
+
+from airflow.configuration import conf
+from airflow.migrations.utils import disable_sqlite_fkeys
+
+# revision identifiers, used by Alembic.
+revision = "0f4c9a2b8e1d"
+down_revision = "c71065a1b14f"
+branch_labels = None
+depends_on = None
+airflow_version = "3.5.0"
+
+
+def _deserialize_data(data):
+ if isinstance(data, str):
+ try:
+ data = json.loads(data)
+ except json.JSONDecodeError:
+ return {}
+ return data.get("__var", data) if isinstance(data, dict) else {}
+
+
+def _extract_dagrun_id(data) -> int | None:
+ dagrun_id = _deserialize_data(data).get("dag_run_id")
+ if dagrun_id is None:
+ return None
+ try:
+ return int(dagrun_id)
+ except (TypeError, ValueError):
+ return None
+
+
+def _backfill_dagrun_id_from_deadline(conn, batch_size: int) -> None:
Review Comment:
Based on the benchmark results, I think it is reasonable to backfill `callback.dagrun_id`.
This is a one-time cost that normalizes existing callbacks to the database-managed relationship. It also lets us remove the read-side fallback to legacy `callback.data` instead of maintaining both representations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org