You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ra...@apache.org on 2026/04/28 13:13:53 UTC
(airflow) branch fix/deadline-migration-0080-downgrade-not-null created (now d0ffd1d889d)
This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a change to branch fix/deadline-migration-0080-downgrade-not-null
in repository https://gitbox.apache.org/repos/asf/airflow.git
at d0ffd1d889d fix: migrate existing deadline rows in migration 0080 upgrade and downgrade
This branch includes the following new commits:
new d0ffd1d889d fix: migrate existing deadline rows in migration 0080 upgrade and downgrade
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(airflow) 01/01: fix: migrate existing deadline rows in migration 0080 upgrade and downgrade
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch fix/deadline-migration-0080-downgrade-not-null
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit d0ffd1d889de19029956acaef1c5fa258246f907
Author: vatsrahul1001 <ra...@gmail.com>
AuthorDate: Tue Apr 28 18:43:32 2026 +0530
fix: migrate existing deadline rows in migration 0080 upgrade and downgrade
Both the upgrade and downgrade paths of migration 0080
(808787349f22 - Modify deadline callback schema) added NOT NULL columns
to the deadline table without first populating them from the existing
data, causing:
* upgrade: NotNullViolation when adding callback JSON NOT NULL to a
non-empty table (existing rows have no value for the new
column).
* downgrade: NotNullViolation on PostgreSQL / silent NULL on MySQL when
adding callback VARCHAR(500) NOT NULL after dropping the
JSON column, crashing the subsequent 0094 upgrade with
json.loads(None).
Fix both paths with the same pattern used by migration 0094:
1. Read the existing data before any schema change.
2. Add the new column(s) as nullable so the DDL succeeds on all
supported databases.
3. Back-fill the column using a typed SA table clause (so each dialect
handles JSON serialisation correctly).
4. Enforce NOT NULL only after every row has a valid value.
Upgrade serialises the old (path, kwargs) pair into the
{"__data__": {"path": ..., "kwargs": ...}, "__classname__": ...,
"__version__": 0} format expected by migration 0094.
Downgrade extracts path/kwargs back from that same JSON envelope
before restoring the VARCHAR callback column.
Made-with: Cursor
---
.../0080_3_1_0_modify_deadline_callback_schema.py | 89 +++++++++++++++++++++-
1 file changed, 87 insertions(+), 2 deletions(-)
diff --git a/airflow-core/src/airflow/migrations/versions/0080_3_1_0_modify_deadline_callback_schema.py b/airflow-core/src/airflow/migrations/versions/0080_3_1_0_modify_deadline_callback_schema.py
index adb3354512d..24b467ce278 100644
--- a/airflow-core/src/airflow/migrations/versions/0080_3_1_0_modify_deadline_callback_schema.py
+++ b/airflow-core/src/airflow/migrations/versions/0080_3_1_0_modify_deadline_callback_schema.py
@@ -27,6 +27,8 @@ Create Date: 2025-07-31 19:35:53.150465
from __future__ import annotations
+import json
+
import sqlalchemy as sa
from alembic import op
@@ -38,17 +40,100 @@ depends_on = None
airflow_version = "3.1.0"
+_ASYNC_CALLBACK_CLASSNAME = "airflow.sdk.definitions.deadline.AsyncCallback"
+
+
def upgrade():
"""Replace deadline table's string callback and JSON callback_kwargs with JSON callback."""
+ conn = op.get_bind()
+
+ # Read the pre-existing string callback and JSON kwargs before the schema
+ # change so we can populate the new JSON callback column.
+ rows = conn.execute(sa.text("SELECT id, callback, callback_kwargs FROM deadline")).fetchall()
+ callback_data: dict = {}
+ for row in rows:
+ path = row[1] or ""
+ kwargs = row[2]
+ if isinstance(kwargs, str):
+ kwargs = json.loads(kwargs) if kwargs else {}
+ elif kwargs is None:
+ kwargs = {}
+ callback_data[row[0]] = {
+ "__data__": {"path": path, "kwargs": kwargs},
+ "__classname__": _ASYNC_CALLBACK_CLASSNAME,
+ "__version__": 0,
+ }
+
+ # Add the new JSON column as nullable first to avoid NotNullViolation when
+ # the table already has rows.
with op.batch_alter_table("deadline", schema=None) as batch_op:
batch_op.drop_column("callback")
batch_op.drop_column("callback_kwargs")
- batch_op.add_column(sa.Column("callback", sa.JSON(), nullable=False))
+ batch_op.add_column(sa.Column("callback", sa.JSON(), nullable=True))
+
+ # Re-populate using a typed table clause so SQLAlchemy handles JSON
+ # serialisation correctly for every supported dialect.
+ deadline_table = sa.table(
+ "deadline",
+ sa.column("id"),
+ sa.column("callback", sa.JSON()),
+ )
+ for row_id, serialized in callback_data.items():
+ conn.execute(
+ sa.update(deadline_table)
+ .where(deadline_table.c.id == row_id)
+ .values(callback=serialized)
+ )
+
+ # Now that every row has a value, enforce the NOT NULL constraint.
+ with op.batch_alter_table("deadline", schema=None) as batch_op:
+ batch_op.alter_column("callback", existing_type=sa.JSON(), nullable=False)
def downgrade():
"""Replace deadline table's JSON callback with string callback and JSON callback_kwargs."""
+ conn = op.get_bind()
+
+ # Read existing JSON callback data before dropping the column so we can
+ # populate the restored VARCHAR callback column. The JSON was written by
+ # the 0094 downgrade in the format:
+ # {"__data__": {"path": "...", "kwargs": {...}}, "__classname__": "...", "__version__": 0}
+ rows = conn.execute(sa.text("SELECT id, callback FROM deadline")).fetchall()
+ callback_data: dict = {}
+ for row in rows:
+ cb = row[1]
+ if cb is None:
+ callback_data[row[0]] = ("", {})
+ continue
+ if isinstance(cb, str):
+ cb = json.loads(cb)
+ cb_inner = cb.get("__data__", cb)
+ callback_data[row[0]] = (cb_inner.get("path", ""), cb_inner.get("kwargs", {}))
+
+ # Add the old columns as nullable first so the schema change succeeds even
+ # when rows already exist (avoids NotNullViolation on PostgreSQL / silent
+ # NULL on MySQL when adding a NOT NULL column without a default value).
with op.batch_alter_table("deadline", schema=None) as batch_op:
batch_op.drop_column("callback")
batch_op.add_column(sa.Column("callback_kwargs", sa.JSON(), nullable=True))
- batch_op.add_column(sa.Column("callback", sa.String(length=500), nullable=False))
+ batch_op.add_column(sa.Column("callback", sa.String(length=500), nullable=True))
+
+ # Re-populate the restored string callback from the data we read above.
+ # Use a typed table clause so SQLAlchemy handles JSON serialisation correctly
+ # for every supported dialect (PostgreSQL, MySQL, SQLite).
+ deadline_table = sa.table(
+ "deadline",
+ sa.column("id"),
+ sa.column("callback", sa.String(500)),
+ sa.column("callback_kwargs", sa.JSON()),
+ )
+ for row_id, (path, kwargs) in callback_data.items():
+ conn.execute(
+ sa.update(deadline_table)
+ .where(deadline_table.c.id == row_id)
+ .values(callback=path, callback_kwargs=kwargs)
+ )
+
+ # Now that every row has a value, enforce the NOT NULL constraint.
+ with op.batch_alter_table("deadline", schema=None) as batch_op:
+ batch_op.alter_column("callback", existing_type=sa.String(length=500), nullable=False)