[ColtenOuO]

Sumarry

Fix the Execution API's XCom sequence-read endpoints so a mapped task/task-group instance that hasn't pushed an XCom yet (still up_for_reschedule, or skipped) doesn't shift the values of every mapped instance after it.

closes: #40321

The bug

get_mapped_xcom_by_index, get_mapped_xcom_by_slice, and head_xcom (in airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py) all built their result by querying XCom rows ordered by map_index and then using the position in that result set as if it were the map_index itself (via SQL OFFSET/LIMIT/slice(), with map_index dropped from the response).

That's correct only when every mapped instance has already pushed an XCom. If one hasn't — e.g. a sensor inside a dynamically mapped @task_group is still up_for_reschedule — there's no row for its map_index, so the query silently returns one fewer row and every later map_index's value shifts down by one position. A downstream task reading mapped_group.output[i] (or iterating over it) can get another instance's value instead of the one it asked for, or hit an IndexError reading past the (now short) end.

This is the same root cause reported in #40321 in 2024 against the Airflow 2.x LazyXComSelectSequence implementation. The AIP-72 rewrite reimplemented the same "position == map_index" assumption in the new task-sdk/execution-API code path instead of fixing it, so it's still present on main today.

Confirming the bug is still there, and that the fix resolves it

Before writing the fix, I reproduced the bug against current main with a test scenario matching the issue: a task mapped over 4 values, where map_index=1 never gets an XCom row (simulating a mapped instance that's still up_for_reschedule and hasn't pushed anything). Querying the slice endpoint for all 4 values returned only 3 — ["f", "o", "b"] — with the value that actually belongs to map_index=3 silently shifted into map_index=1's position. The HEAD endpoint's count also undercounted (3 instead of 4). This matches exactly what the issue reports: values read by a downstream task line up with the wrong map_index, and the reported length is wrong too.

Two of the existing tests in test_xcoms.py (test_xcom_get_with_offset, test_xcom_get_with_slice) already built this exact gapped scenario, but asserted the compacted 3-item result as the expected outcome — meaning the buggy behavior was pinned as correct by the test suite rather than being an untested gap.

After the fix, I reran the same scenario: the slice endpoint now returns the full 4-item ["f", None, "o", "b"], with map_index=3's value correctly in position 3, and HEAD's count correctly reports 4. I updated the two pinning tests to assert this corrected behavior and added test_xcom_get_with_slice_and_count_unfinished_mapped_task as an explicit regression test for the issue's scenario (a task instance that never pushed an XCom at all, as opposed to one whose pushed value happened to be None — those are different situations that both need to resolve correctly).

The fix

Added _get_mapped_length(), which resolves the logical number of mapped instances for (dag_id, run_id, task_id) from the TaskInstance table (every mapped instance gets a row at expansion time, whether or not it has run yet — see TaskMap.expand_mapped_task), independent of how many XCom rows exist.

When a task is mapped in this run, the three endpoints now key results by the actual map_index instead of result position, filling in None for any map_index that has no XCom row yet:

• get_mapped_xcom_by_index: offset resolves against the logical 0..mapped_length-1 range; a valid-but-missing index returns None (200), not a 404 — only an out-of-range index 404s.
• get_mapped_xcom_by_slice: builds a map_index -> value dict, fills gaps with None, then applies the slice with plain Python slicing.
• head_xcom: reports the logical mapped length in Content-Range, not the XCom row count.

Two cases are deliberately left on the old (pre-fix) behavior, since there's no well-defined "logical length" for them:

• include_prior_dates=True on the slice endpoint can span multiple dag runs, each with its own (potentially different) mapped length.
• An unmapped task (map_index=-1, no expansion) has no mapped-length concept at all — it falls through to the original row-position logic unchanged.

get_xcom's offset query-parameter branch (a separate, older code path) is intentionally untouched — no task-sdk client currently calls it with offset set, and it has its own pre-existing "skip None values" semantics that aren't part of this bug.

Blast radius: what else calls these endpoints, and what changes for callers

I checked who actually consumes these three routes and what the response-shape change means for them, since this touches a fairly central part of the Execution API.

Consumers. task-sdk's LazyXComSequence (task-sdk/src/airflow/sdk/execution_time/lazy_sequence.py) is the only real consumer, reached through XComOperations in task-sdk/src/airflow/sdk/api/client.py. I grepped the rest of the repo (providers/, the UI, everywhere else) for these route paths and the relevant symbol names and found no other caller — nothing outside task-sdk talks to these three endpoints directly. client.py itself is a thin pass-through and doesn't assume anything about the old semantics, so it needed no changes.

User-visible behavior change. This is the one worth calling out explicitly: a mapped task's .output used to silently compact away gaps when iterated, sliced, or expanded over downstream (for v in t.output, .expand(x=t.output), etc.) — a missing map_index just meant the list was shorter than the actual mapped count, with everything after it shifted into the wrong position. After this fix, .output is the correct length and gaps show up explicitly as None at their true position. This is the intended correctness fix, but it's a behavior change any DAG author relying (knowingly or not) on the old compaction would notice — e.g. .expand(x=t.output) now creates an instance with x=None for the gap instead of one fewer instance. Flagging this for review since it's a semantic change, not just a bugfix that's invisible from the outside.

Performance. The original version of this fix always fetched every XCom row for the task into a Python dict once a task was detected as mapped, even when there was no gap at all — regressing the existing SQL-side OFFSET/LIMIT/slice() pagination for the common case. Since XCom values are stored in an unbounded JSON column (only the count of mapped instances is capped, by core.max_map_length, default 1024), that could mean loading a large number of large payloads into API-server memory for a single-item slice request. Fixed by adding one cheap COUNT query to detect whether a gap actually exists before falling into the full-fetch path — the common case (no gaps) keeps the original SQL pagination; only a genuinely sparse sequence pays the cost of loading every row, which is unavoidable since placing values at their true map_index requires knowing all of them. Added test_xcom_get_with_slice_mapped_task_without_gap_uses_sql_pagination to cover the no-gap path explicitly, since the existing gap-scenario tests wouldn't have caught a regression here.

---

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (Sonnet 5)

[potiuk]

Hello @ColtenOuO - thank you for your contributions to Apache Airflow!

The Airflow community has introduced a limit of 5 open pull requests at a time for contributors without write access to the repository. You currently have 19 open pull requests, so - as a one-time step of introducing the limit - we closed the ones where maintainers have not engaged yet:

• [AIP-94] Add airflowctl Tasks List Command #70472 - [AIP-94] Add airflowctl Tasks List Command
• Let HookToolset retry a failed tool call instead of failing the task #71853 - Let HookToolset retry a failed tool call instead of failing the task
• Fix scheduler crash when the API server response is a redirect #71962 - Fix scheduler crash when the API server response is a redirect
• Fix manually triggered runs blocking a scheduled run at the same logical_date #72003 - Fix manually triggered runs blocking a scheduled run at the same logical_date
• Fix mismatched XCom values when a dynamically mapped task has a gap #72007 - Fix mismatched XCom values when a dynamically mapped task has a gap
• Refactor schema version checks to share target-branch and worktree logic #72070 - Refactor schema version checks to share target-branch and worktree logic
• Document custom OTel metrics endpoint path support #72073 - Document custom OTel metrics endpoint path support
• Batch XCom lookups when resolving a mapped task's expand() arguments #72107 - Batch XCom lookups when resolving a mapped task's expand() arguments
• Remove redundant full-Dag TaskInstance query from the execution API task-states endpoint #72270 - Remove redundant full-Dag TaskInstance query from the execution API task-states endpoint
• Fix asset dependency graph dropping outlets for tasks that both consume and produce #72272 - Fix asset dependency graph dropping outlets for tasks that both consume and produce
• Make supervisor split-line stdout test actually cover the split in Task SDK #72585 - Make supervisor split-line stdout test actually cover the split in Task SDK
• Fix GET /assets/{asset_id} returning 500 when asset events share a timestamp #72599 - Fix GET /assets/{asset_id} returning 500 when asset events share a timestamp
• Show detailed API error messages in UI toasters #72831 - Show detailed API error messages in UI toasters

These pull requests stay open because maintainers are already engaged in them - they count towards your limit:

• Fix output type loss in AgentOperator's human-in-the-loop review #70132 - Fix output type loss in AgentOperator's human-in-the-loop review
• Decode base64 for bytes-typed HookToolset parameters #70433 - Decode base64 for bytes-typed HookToolset parameters
• Allow providers to ship testing Dags separately from example Dags #70541 - Allow providers to ship testing Dags separately from example Dags
• Reject unknown update_mask fields instead of silently ignoring them #71003 - Reject unknown update_mask fields instead of silently ignoring them
• Speed up clearing downstream tasks on large Dags #71203 - Speed up clearing downstream tasks on large Dags
• Add deadline alert coverage to Dag endpoint tests #72296 - Add deadline alert coverage to Dag endpoint tests

This is not a judgement of you or of your changes. We never told contributors before that opening many pull requests at once was a problem, so there is nothing to feel bad about - and nothing is lost: your branches, commits and the review history stay where they are.

What we ask you to do is to make your first prioritization decision: choose which of the pull requests above matter most to you, and reopen them (up to 5 open at a time, including the ones still open) with the "Reopen pull request" button or gh pr reopen <PR_NUMBER> --repo apache/airflow. Reopen the ones you are ready to follow through - keep them rebased, respond to review comments and fix failing checks.

While your pull requests are waiting for review, the most valuable thing you can do is help in other ways - reviewing other contributors' pull requests, helping with issues, and taking part in the discussions on the devlist and Slack.

Why we introduced the limit, what it means for you and how to reopen or restore a pull request is explained in https://github.com/apache/airflow/blob/main/contributing-docs/32_open_pull_request_limit.rst.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting