Skip to content

Pin loop definitions and carry region coordinates through task execution - #74345

Merged
ashb merged 1 commit into
task-loops-stack-3from
task-loops-stack-4
Oct 10, 2026
Merged

ashb merged 1 commit into
task-loops-stack-3from
task-loops-stack-4

Conversation

@ashb

@ashb ashb commented Oct 6, 2026 •

Copy link
Copy Markdown
Member

A task in a loop or mapped region is identified by (region_id, region_index),
but running code, the Execution API and log readers still address it by the
public map_index. If one value served both, a loop pass would look like a mapped
expansion. We derive the public index from the Dag definition pinned to the try
instead of storing it: a task exposes its region index as map_index only when
its pinned definition expands, and -1 otherwise. Ordinary mapped tasks therefore
keep reporting exactly what they did before.

The scheduler and API server must understand loops without importing user code,
so the serialized Dag carries a loop group definition they can read.

The Execution API and the supervisor schema take the region fields as optional,
versioned fields so that older task SDKs keep working against a newer API
server. Lookups of earlier runs refuse loop producers, because an iteration of
one run has no counterpart to resolve in another run.

A log path and a remote log id have to come out the same for the process that
writes the log and for every reader that fetches it later, retired attempts
included, and one loop pass must not overwrite the logs of the pass before it.
Building both from one prepared, batched context gives workers, the triggerer
and log readers the same answer without a query per task. Only loop tasks gain
a pass position; ordinary mapped tasks keep their existing path, so logs already
written stay readable.


Was generative AI tooling used to co-author this PR?
  • Yes (please specify the tool below)

  • Read the 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) is needed.
  • When adding dependency, check compliance with the ASF 3rd Party License Policy.
  • For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

Comment thread task-sdk/src/airflow/sdk/execution_time/task_runner.py
Comment thread airflow-core/src/airflow/utils/log/task_log_address.py Outdated
Comment thread airflow-core/src/airflow/executors/workloads/task.py
Comment thread go-sdk/airflow/spec.gen.go Outdated
Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py Outdated
Comment thread task-sdk/src/airflow/sdk/execution_time/schema/versions/v2026_10_30.py Outdated
Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py Outdated
Comment thread airflow-core/src/airflow/models/task_coordinates.py Outdated
Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py Outdated
Comment thread airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_xcoms.py Outdated
@ashb
ashb force-pushed the task-loops-stack-4 branch from 3b98054 to adce91a Compare October 7, 2026 13:57
@ashb
ashb removed this pull request from stack #74339 October 7, 2026 15:20
@ashb
ashb force-pushed the task-loops-stack-4 branch from adce91a to 45978c7 Compare October 7, 2026 15:22
@ashb
ashb added this pull request to stack #74410 October 7, 2026 15:22
@ashb
ashb force-pushed the task-loops-stack-4 branch from 45978c7 to 6829d3c Compare October 7, 2026 20:57
@ashb
ashb force-pushed the task-loops-stack-4 branch from 8b8d409 to 94a81e0 Compare October 8, 2026 13:48
@ashb
ashb force-pushed the task-loops-stack-4 branch from 94a81e0 to 7774d1b Compare October 8, 2026 13:53
@ashb
ashb marked this pull request as ready for review October 8, 2026 13:58
Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py Outdated
Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py Outdated
Comment thread providers/edge3/src/airflow/providers/edge3/worker_api/routes/logs.py Outdated
Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/xcoms.py Outdated
Comment thread airflow-core/src/airflow/jobs/scheduler_job_runner.py
Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py Outdated
Comment thread airflow-core/src/airflow/api_fastapi/execution_api/datamodels/taskinstance.py Outdated
Comment thread airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py Outdated
@ashb
ashb force-pushed the task-loops-stack-4 branch from 07542b1 to fb403e9 Compare October 8, 2026 20:45
@ashb
ashb force-pushed the task-loops-stack-4 branch from fb403e9 to 4741eef Compare October 9, 2026 10:32
@ashb
ashb force-pushed the task-loops-stack-4 branch 2 times, most recently from 2f7bed1 to ca94467 Compare October 9, 2026 16:28
@ashb
ashb force-pushed the task-loops-stack-4 branch 2 times, most recently from 15795c7 to b3f9aa1 Compare October 9, 2026 20:46
@ashb
ashb force-pushed the task-loops-stack-4 branch 2 times, most recently from b93e57d to fc3aa41 Compare October 10, 2026 07:05
A task in a loop or mapped region is identified by (region_id, region_index),
but running code, the Execution API and log readers still address it by the
public map_index. If one value served both, a loop pass would look like a mapped
expansion. We derive the public index from the Dag definition pinned to the try
instead of storing it: a task exposes its region index as map_index only when
its pinned definition expands, and -1 otherwise. Ordinary mapped tasks therefore
keep reporting exactly what they did before.

The scheduler and API server must understand loops without importing user code,
so the serialized Dag carries a loop group definition they can read.

The Execution API and the supervisor schema take the region fields as optional,
versioned fields so that older task SDKs keep working against a newer API
server. Lookups of earlier runs refuse loop producers, because an iteration of
one run has no counterpart to resolve in another run.

A log path and a remote log id have to come out the same for the process that
writes the log and for every reader that fetches it later, retired attempts
included, and one loop pass must not overwrite the logs of the pass before it.
Building both from one prepared, batched context gives workers, the triggerer
and log readers the same answer without a query per task. Only loop tasks gain
a pass position; ordinary mapped tasks keep their existing path, so logs already
written stay readable.
@ashb
ashb force-pushed the task-loops-stack-4 branch from fc3aa41 to 2fd9c2a Compare October 10, 2026 16:18
@ashb
ashb merged commit 9bc8cad into main Oct 10, 2026
217 of 218 checks passed
@ashb
ashb deleted the task-loops-stack-4 branch October 10, 2026 22:49
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants