Visitar URL original
SerializedDAG.partial_subset modifies the original DAG when it contains a mapped TaskGroup (ReferenceError with nested TaskGroups) · Issue #74453 · apache/airflow · GitHub
Skip to content

SerializedDAG.partial_subset modifies the original DAG when it contains a mapped TaskGroup (ReferenceError with nested TaskGroups) #74453

Open
#74541
Open
SerializedDAG.partial_subset modifies the original DAG when it contains a mapped TaskGroup (ReferenceError with nested TaskGroups)#74453
#74541
Labels
kind:bugThis is a clearly a bugneeds-triagelabel for new issues that we didn't triage yet

Description

@jgoedeke
jgoedeke
opened on Oct 8, 2026

Under which category would you file this issue?

Airflow Core

Apache Airflow version

3.3.2

What happened and how to reproduce it?

In SerializedDAG.partial_subset() (https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/serialization/definitions/dag.py#L356-L385), filter_task_group() shallow-copies each task group and then deep-copies the attributes listed in type(group).__slots__:

copied = copy.copy(group)
...
for attr in type(group).__slots__:
    value = getattr(group, attr)
    value = copy.deepcopy(value, memo)
    object.__setattr__(copied, attr, value)

On 3.3.2:

SerializedTaskGroup.__slots__       = ('_group_id', 'group_display_name', 'prefix_group_id', 'parent_group', 'dag', 'tooltip', 'doc_md', 'default_args', 'ui_color', 'ui_fgcolor', 'children', 'upstream_group_ids', 'downstream_group_ids', 'upstream_task_ids', 'downstream_task_ids')
SerializedMappedTaskGroup.__slots__ = ('_expand_input',)

For a SerializedMappedTaskGroup, children is not in type(group).__slots__, so the copy keeps the original children dict (copied.children is group.children is True for the mapped group and False for the root group and the nested SerializedTaskGroup). filter_task_group() then assigns the subset's tasks and copied child groups into that dict, so after one call the original DAG's mapped group contains the subset's task objects and the subset's copy of the nested group.

Behaviour after one partial_subset() call + gc.collect() on 3.3.2:

DAG shape Original DAG modified Original tasks with dead task_group weakref
mapped group containing a nested group yes tasks of the nested group
mapped group, no nested group yes none
unmapped group containing a nested group no none

Reproduction (no database needed):

import gc
import tracemalloc

from airflow.sdk import DAG, task, task_group
from airflow.serialization.serialized_objects import DagSerialization


@task
def files():
    return [1, 2]


@task
def prepare(x):
    return x


@task
def convert(x):
    return x


@task_group
def inner(x):
    a = convert.override(task_id='a')(x)
    b = convert.override(task_id='b')(a)
    return b


@task
def select(x):
    return x


@task_group
def outer(x):
    p = prepare(x)
    c = convert(p)
    i = inner(c)
    return select(i)


with DAG('repro') as dag:
    outer.expand(x=files())

sdag = DagSerialization.from_dict(DagSerialization.to_dict(dag))

tracemalloc.start()
gc.collect()
base = tracemalloc.get_traced_memory()[0]
for i in range(1, 201):
    sub = sdag.partial_subset(
        task_ids=['outer.prepare'], include_downstream=True, include_upstream=False
    )
    del sub
    if i in (1, 2, 5, 50, 200):
        gc.collect()
        cur = tracemalloc.get_traced_memory()[0]
        inner_group = sdag.task_group.children['outer'].children.get('outer.inner')
        try:
            tg = sdag.task_dict['outer.inner.a'].task_group.group_id
        except ReferenceError as e:
            tg = f'ReferenceError: {e}'
        print(
            f'iter={i} mem_delta={(cur - base) / 1024:.0f}KiB '
            f'inner_id={id(inner_group):x} task_group(a)={tg}'
        )

sdag.task_group.topological_sort()
print('topological_sort ok')

Output on 3.3.2 (Python 3.11; ids and KiB values vary between runs):

iter=1 mem_delta=23KiB inner_id=70952194a410 task_group(a)=ReferenceError: weakly-referenced object no longer exists
iter=2 mem_delta=61KiB inner_id=709521948890 task_group(a)=ReferenceError: weakly-referenced object no longer exists
iter=5 mem_delta=132KiB inner_id=7095219a8ec0 task_group(a)=ReferenceError: weakly-referenced object no longer exists
iter=50 mem_delta=937KiB inner_id=7095217bd650 task_group(a)=ReferenceError: weakly-referenced object no longer exists
Traceback (most recent call last):
  File "/tmp/issue_exact.py", line 52, in <module>
    sub = sdag.partial_subset(
  File ".../airflow/serialization/definitions/dag.py", line 382, in partial_subset
    object.__setattr__(dag, "task_group", filter_task_group(self.task_group, None))
  File ".../airflow/serialization/definitions/dag.py", line 374, in filter_task_group
    filtered_child = filter_task_group(child, proxy)
  File ".../airflow/serialization/definitions/dag.py", line 374, in filter_task_group
    filtered_child = filter_task_group(child, proxy)
  File ".../airflow/serialization/definitions/dag.py", line 363, in filter_task_group
    value = copy.deepcopy(value, memo)
  [... repeated copy.deepcopy frames ...]
  File ".../attr/_make.py", line 1034, in slots_getstate
    return {name: getattr(self, name) for name in state_attr_names}
RecursionError: maximum recursion depth exceeded

The outer.inner group object in the original sdag is a different object at each sampled iteration (inner_id).

The Grid code path produces the same ReferenceError after a single partial_subset() call (same sdag construction as above):

import gc
from airflow.api_fastapi.core_api.services.ui.grid import _find_aggregates

list(_find_aggregates(node=sdag.task_group, parent_node=None, ti_details={}))  # ok (8 nodes)
sdag.partial_subset(task_ids=['outer.prepare'], include_downstream=True, include_upstream=False)
gc.collect()
list(_find_aggregates(node=sdag.task_group, parent_node=None, ti_details={}))  # ReferenceError
  File ".../airflow/api_fastapi/core_api/services/ui/grid.py", line 178, in _find_aggregates
  File ".../airflow/api_fastapi/core_api/services/ui/task_group.py", line 40, in <lambda>
  File ".../airflow/serialization/definitions/taskgroup.py", line 257, in topological_sort
  File ".../airflow/serialization/definitions/taskgroup.py", line 291, in _project_child_deps
    anc_idx = id_to_idx.get(tg.node_id)
ReferenceError: weakly-referenced object no longer exists

In 3.3.2, partial_subset() is called by the API server in POST .../clearTaskInstances (routes/public/task_instances.py), routes/ui/grid.py (when root is given) and routes/ui/structure.py.

In our deployment (DAG with a mapped task group containing a nested task group), the API server logged the following for the Grid TI summaries endpoint (GET /ui/grid/ti_summaries/{dag_id}):

2026-10-08T06:24:14.205651359Z [2026-10-08T06:24:14.203182Z] {httptools_impl.py:427} ERROR - Exception in ASGI application
2026-10-08T06:24:14.205668653Z
2026-10-08T06:24:14.205672382Z Traceback (most recent call last):
2026-10-08T06:24:14.205675708Z   File "/home/airflow/.local/lib/python3.11/site-packages/uvicorn/protocols/http/httptools_impl.py", line 422, in run_asgi
2026-10-08T06:24:14.205679291Z     result = await app(  # type: ignore[func-returns-value]
2026-10-08T06:24:14.205682579Z              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.205698870Z   File "/home/airflow/.local/lib/python3.11/site-packages/fastapi/applications.py", line 1159, in __call__
2026-10-08T06:24:14.205703093Z     await super().__call__(scope, receive, send)
2026-10-08T06:24:14.205706324Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/applications.py", line 96, in __call__
2026-10-08T06:24:14.205709670Z     await self.middleware_stack(scope, receive, send)
2026-10-08T06:24:14.205712846Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 186, in __call__
2026-10-08T06:24:14.205716164Z     raise exc
2026-10-08T06:24:14.205719346Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/errors.py", line 164, in __call__
2026-10-08T06:24:14.205722692Z     await self.app(scope, receive, _send)
2026-10-08T06:24:14.205725973Z   File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/common/http_access_log.py", line 109, in __call__
2026-10-08T06:24:14.205729477Z     await self.app(scope, receive, capture_send)
2026-10-08T06:24:14.205732674Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/gzip.py", line 78, in __call__
2026-10-08T06:24:14.205736058Z     await responder(scope, receive, send)
2026-10-08T06:24:14.205739348Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/gzip.py", line 103, in __call__
2026-10-08T06:24:14.205742621Z     await self.app(scope, receive, self.send_with_compression)
2026-10-08T06:24:14.205745830Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/base.py", line 198, in __call__
2026-10-08T06:24:14.205749223Z     raise app_exc
2026-10-08T06:24:14.205752294Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/base.py", line 144, in coro
2026-10-08T06:24:14.205755542Z     await self.app(scope, receive_or_disconnect, send_no_error)
2026-10-08T06:24:14.205855421Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/middleware/exceptions.py", line 63, in __call__
2026-10-08T06:24:14.205859773Z     await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
2026-10-08T06:24:14.205863704Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
2026-10-08T06:24:14.205867030Z     raise exc
2026-10-08T06:24:14.205870052Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
2026-10-08T06:24:14.205873264Z     await app(scope, receive, sender)
2026-10-08T06:24:14.205876311Z   File "/home/airflow/.local/lib/python3.11/site-packages/fastapi/middleware/asyncexitstack.py", line 18, in __call__
2026-10-08T06:24:14.205879575Z     await self.app(scope, receive, send)
2026-10-08T06:24:14.205883039Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/routing.py", line 670, in __call__
2026-10-08T06:24:14.205886261Z     await self.middleware_stack(scope, receive, send)
2026-10-08T06:24:14.205889376Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/routing.py", line 690, in app
2026-10-08T06:24:14.205892552Z     await route.handle(scope, receive, send)
2026-10-08T06:24:14.205895615Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/routing.py", line 280, in handle
2026-10-08T06:24:14.205898803Z     await self.app(scope, receive, send)
2026-10-08T06:24:14.205901916Z   File "/home/airflow/.local/lib/python3.11/site-packages/fastapi/routing.py", line 134, in app
2026-10-08T06:24:14.205905100Z     await wrap_app_handling_exceptions(app, request)(scope, receive, send)
2026-10-08T06:24:14.205908277Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 53, in wrapped_app
2026-10-08T06:24:14.205911864Z     raise exc
2026-10-08T06:24:14.205914824Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_exception_handler.py", line 42, in wrapped_app
2026-10-08T06:24:14.205918072Z     await app(scope, receive, sender)
2026-10-08T06:24:14.206012036Z   File "/home/airflow/.local/lib/python3.11/site-packages/fastapi/routing.py", line 121, in app
2026-10-08T06:24:14.206016255Z     await response(scope, receive, send)
2026-10-08T06:24:14.206019426Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/responses.py", line 273, in __call__
2026-10-08T06:24:14.206022767Z     async with create_collapsing_task_group() as task_group:
2026-10-08T06:24:14.206025901Z   File "/usr/python/lib/python3.11/contextlib.py", line 231, in __aexit__
2026-10-08T06:24:14.206029113Z     await self.gen.athrow(typ, value, traceback)
2026-10-08T06:24:14.206032247Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/_utils.py", line 93, in create_collapsing_task_group
2026-10-08T06:24:14.206035548Z     raise exc from exc.__cause__ or context
2026-10-08T06:24:14.206042854Z   File "/home/airflow/.local/lib/python3.11/site-packages/anyio/_core/_tasks.py", line 327, in _run_coro
2026-10-08T06:24:14.206046298Z     retval = await self._coro
2026-10-08T06:24:14.206049457Z              ^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.206052970Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/responses.py", line 276, in wrap
2026-10-08T06:24:14.206056357Z     await func()
2026-10-08T06:24:14.206059422Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/responses.py", line 250, in stream_response
2026-10-08T06:24:14.206062753Z     async for chunk in self.body_iterator:
2026-10-08T06:24:14.206065871Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/concurrency.py", line 57, in iterate_in_threadpool
2026-10-08T06:24:14.206069210Z     yield await anyio.to_thread.run_sync(_next, as_iterator)
2026-10-08T06:24:14.206072302Z           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.206075401Z   File "/home/airflow/.local/lib/python3.11/site-packages/anyio/to_thread.py", line 65, in run_sync
2026-10-08T06:24:14.206173524Z     return await get_async_backend().run_sync_in_worker_thread(
2026-10-08T06:24:14.206177640Z            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.206180835Z   File "/home/airflow/.local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 2706, in run_sync_in_worker_thread
2026-10-08T06:24:14.206184296Z     return await future
2026-10-08T06:24:14.206187506Z            ^^^^^^^^^^^^
2026-10-08T06:24:14.206190569Z   File "/home/airflow/.local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 1100, in run
2026-10-08T06:24:14.206193887Z     result = context.run(func, *args)
2026-10-08T06:24:14.206196966Z              ^^^^^^^^^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.206200114Z   File "/home/airflow/.local/lib/python3.11/site-packages/starlette/concurrency.py", line 46, in _next
2026-10-08T06:24:14.206203466Z     return next(iterator)
2026-10-08T06:24:14.206206560Z            ^^^^^^^^^^^^^^
2026-10-08T06:24:14.206209779Z   File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/routes/ui/grid.py", line 516, in _generate
2026-10-08T06:24:14.206213208Z     summary = _build_ti_summaries(
2026-10-08T06:24:14.206216620Z               ^^^^^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.206219666Z   File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/routes/ui/grid.py", line 430, in _build_ti_summaries
2026-10-08T06:24:14.206389195Z     nodes = list(get_node_summaries())
2026-10-08T06:24:14.206393373Z             ^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.206396537Z   File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/routes/ui/grid.py", line 407, in get_node_summaries
2026-10-08T06:24:14.206399875Z     for node, _ in _find_aggregates(
2026-10-08T06:24:14.206406086Z   File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/services/ui/grid.py", line 179, in _find_aggregates
2026-10-08T06:24:14.206409664Z     for child_node, child_summary in _find_aggregates(
2026-10-08T06:24:14.206413191Z   File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/services/ui/grid.py", line 178, in _find_aggregates
2026-10-08T06:24:14.206416626Z     for child in get_task_group_children_getter()(node, group_dict):
2026-10-08T06:24:14.206419866Z                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.206422946Z   File "/home/airflow/.local/lib/python3.11/site-packages/airflow/api_fastapi/core_api/services/ui/task_group.py", line 40, in <lambda>
2026-10-08T06:24:14.206426589Z     return lambda task_group, group_dict=None: task_group.topological_sort(group_dict=group_dict)
2026-10-08T06:24:14.206429793Z                                                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.206433050Z   File "/home/airflow/.local/lib/python3.11/site-packages/airflow/serialization/definitions/taskgroup.py", line 257, in topological_sort
2026-10-08T06:24:14.206436421Z     deps = self._project_child_deps(i, child, id_to_idx, group_dict)
2026-10-08T06:24:14.206439499Z            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
2026-10-08T06:24:14.206442537Z   File "/home/airflow/.local/lib/python3.11/site-packages/airflow/serialization/definitions/taskgroup.py", line 291, in _project_child_deps
2026-10-08T06:24:14.206445791Z     anc_idx = id_to_idx.get(tg.node_id)
2026-10-08T06:24:14.206448813Z                             ^^^^^^^^^^
2026-10-08T06:24:14.206451913Z ReferenceError: weakly-referenced object no longer exists

What you think should happen instead?

partial_subset() should not modify the DAG it is called on.

Replacing the loop with one that includes inherited attrs fields:

for attr in (field.name for field in attrs.fields(type(group))):

gives, for the same script on 3.3.2 (and the _find_aggregates walk still returns 8 nodes after 200 partial_subset() calls):

iter=1 mem_delta=4KiB inner_id=7841eb5c0260 task_group(a)=outer.inner
iter=2 mem_delta=4KiB inner_id=7841eb5c0260 task_group(a)=outer.inner
iter=5 mem_delta=5KiB inner_id=7841eb5c0260 task_group(a)=outer.inner
iter=50 mem_delta=13KiB inner_id=7841eb5c0260 task_group(a)=outer.inner
iter=200 mem_delta=28KiB inner_id=7841eb5c0260 task_group(a)=outer.inner
topological_sort ok

After applying this change to our 3.3.2 deployment, the ReferenceError no longer appears in the API server logs.

Operating System

No response

Deployment

Other Docker-based deployment

Apache Airflow Provider(s)

No response

Versions of Apache Airflow Providers

No response

Official Helm Chart version

Not Applicable

Kubernetes Version

No response

Helm Chart configuration

No response

Docker Image customizations

  • Image based on apache/airflow:slim-3.3.2-python3.11
  • CeleryExecutor, Redis 7.2, PostgreSQL 13

Anything else?

Possibly related: #58499 (ReferenceError in API Server during Dynamic Mapping of Nested TaskGroups), closed 2025-11-20.

Created with Github Copilot (Claude Opus 5.5) reviewed by @jgoedeke.

Are you willing to submit PR?

  • Yes I am willing to submit a PR!

Code of Conduct

Activity

  1. jgoedeke
    added
    kind:bugThis is a clearly a bug
    needs-triagelabel for new issues that we didn't triage yet
    on Oct 8, 2026

  2. aryan9948 commented on Oct 8, 2026

    @aryan9948
    aryan9948
    on Oct 8, 2026

    Hi @jgoedeke, I'd like to work on this issue. I've looked through the partial_subset() behavior and the SerializedMappedTaskGroup inheritance/slots issue described here. I'd like to add a regression test for the shared children state and implement the fix using the attrs fields so inherited attributes are copied as well. Could you assign this issue to me?

  3. jgoedeke commented on Oct 9, 2026

    @jgoedeke
    jgoedeke
    on Oct 9, 2026
    ContributorAuthor

    @aryan9948 a PR is very welcome! Issues are usually not assigned anymore, see Assignment Policy

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    kind:bugThis is a clearly a bugneeds-triagelabel for new issues that we didn't triage yet

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions