This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 2bb5d016b34 Fixing bad cadwyn migration for upstream map indexes
(#52797)
2bb5d016b34 is described below
commit 2bb5d016b3404206e7d49ed5371f2f11e07aa537
Author: Amogh Desai <[email protected]>
AuthorDate: Thu Jul 3 22:27:50 2025 +0530
Fixing bad cadwyn migration for upstream map indexes (#52797)
---
.../airflow/api_fastapi/execution_api/versions/v2025_05_20.py | 9 +++++----
.../execution_api/versions/v2025_04_28/test_task_instances.py | 9 ++++++++-
2 files changed, 13 insertions(+), 5 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_20.py
b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_20.py
index d671c97dafd..8ea59844d98 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_20.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_20.py
@@ -28,7 +28,7 @@ class DowngradeUpstreamMapIndexes(VersionChange):
description = __doc__
instructions_to_migrate_to_previous_version = (
- schema(TIRunContext).field("upstream_map_indexes").had(type=dict[str,
int] | None), # type: ignore
+ schema(TIRunContext).field("upstream_map_indexes").had(type=dict[str,
int | None] | None),
)
@convert_response_to_previous_version_for(TIRunContext) # type:
ignore[arg-type]
@@ -40,13 +40,14 @@ class DowngradeUpstreamMapIndexes(VersionChange):
"""
resp = response.body.get("upstream_map_indexes")
if isinstance(resp, dict):
- downgraded = {}
+ downgraded: dict[str, int | list | None] = {}
for k, v in resp.items():
if isinstance(v, int):
downgraded[k] = v
elif isinstance(v, list) and v and all(isinstance(i, int) for
i in v):
downgraded[k] = v[0]
else:
- # for cases like None, make it -1
- downgraded[k] = -1
+ # Keep values like None as is — the Task SDK expects them
unchanged during mapped task expansion,
+ # and modifying them can cause unexpected failures.
+ downgraded[k] = None
response.body["upstream_map_indexes"] = downgraded
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_04_28/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_04_28/test_task_instances.py
index 579559edc0a..b5ef9a3ce5e 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_04_28/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_04_28/test_task_instances.py
@@ -62,9 +62,16 @@ class TestTIUpdateState:
{"task_a": 3, "task_b": 9},
id="list of ints",
),
+ pytest.param(
+ [
+ ("task_a", None),
+ ],
+ {"task_a": None},
+ id="task has no upstreams",
+ ),
pytest.param(
[("task_a", None), ("task_b", [6, 7]), ("task_c", 2)],
- {"task_a": -1, "task_b": 6, "task_c": 2},
+ {"task_a": None, "task_b": 6, "task_c": 2},
id="mixed types",
),
],
