[PR] Fix SerializedDAG.partial_subset mutating the original DAG when it contains a mapped TaskGroup [airflow]

["liwenjie200543 (via GitHub)" <gi...@apache.org>]

liwenjie200543 opened a new pull request, #74541:
URL: https://github.com/apache/airflow/pull/74541

   closes: #74453
   
   ##### Was generative AI tooling used to co-author this PR?
   
   - [X] Yes (please specify the tool below)
   
   Generated-by: ZCode CLI (AI coding agent), following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
   
   ---
   
   ### Summary
   
   `SerializedDAG.partial_subset()` mutated the DAG it was called on whenever the DAG contained a **mapped** TaskGroup. `filter_task_group()` deep-copies only the attributes listed in `type(group).__slots__`, and `SerializedMappedTaskGroup` declares just `_expand_input` in its own `__slots__` — so `children` was never copied and stayed shared with the original group. The subset's tasks and child groups were then assigned into that shared dict, so after a single call the original DAG's mapped group contained the subset's objects, and tasks of nested groups ended up with dead `task_group` weakrefs (`ReferenceError: weakly-referenced object no longer exists` in the Grid/structure endpoints and in `topological_sort()`).
   
   The fix replaces the `__slots__` loop with one that iterates all inherited attrs fields:
   
   ```python
   for attr in (field.name for field in attrs.fields(type(group))):
   ```
   
   `attrs` is already imported in this module, and the loop semantics for non-mapped groups are unchanged (they enumerate the same fields their `__slots__` list).
   
   ### What was verified
   
   - Reproduction from #74453 (mapped group containing a nested group, no database needed) against a clean `apache-airflow==3.3.2` install, Python 3.11:
     - before the fix: after one `partial_subset()` call + `gc.collect()`, the original DAG's `outer.inner` group object is replaced (`id()` changes), `sdag.task_dict["outer.inner.a"].task_group.group_id` raises `ReferenceError`;
     - after the fix: the original group object is untouched, weakrefs stay alive, `task_group.group_id` returns `outer.inner`.
   - Added a regression test in `airflow-core/tests/unit/serialization/test_dag_serialization.py` covering exactly this DAG shape (mapped group + nested group) asserting the original group object identity is preserved and the nested tasks' `task_group` stays resolvable.
   - `ruff format` and `ruff check` pass on both touched files.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] Fix SerializedDAG.partial_subset mutating the original DAG when it contains a mapped TaskGroup [airflow]

["SEPURI-SAI-KRISHNA (via GitHub)" <gi...@apache.org>]

SEPURI-SAI-KRISHNA commented on PR #74541:
URL: https://github.com/apache/airflow/pull/74541#issuecomment-6291285139

   Confirming this fix independently, and adding three things the issue does not cover.
   
   Verified on `5a64dfcc78`; `partial_subset` is byte for byte unchanged at current main `12df3270b8`, so this is still live.
   
   **1. Loop task groups are affected too.** `SerializedLoopTaskGroup` has the same slots shape as the mapped one, so the same aliasing applies:
   
   ```
   SerializedLoopTaskGroup.__slots__ = ('max_iterations', 'has_until', 'terminal_task_id', 'gate_task_id')
   ```
   
   A plain `TaskGroup` nested inside a `create_loop` body crashes through a different path, with no `include_upstream` or `include_downstream` set:
   
   ```text
   routes/public/task_instances.py:1323  post_clear_task_instances
   services/public/task_coordinates.py:100  task_coordinate_responses
   models/task_coordinates.py:191  prefetch_regions
   models/task_coordinates.py:213  _is_in_loop
   models/task_coordinates.py:86  enclosing_loop
   ReferenceError: weakly-referenced object no longer exists
   ```
   
   **2. Group edges are stripped, not only `children` replaced.** Because the upstream and downstream sets are aliased as well, the `intersection_update` pass at the end of `partial_subset` narrows the original group's edges. This happens even when the subset excludes the group entirely:
   
   ```text
   subset of an unrelated task, mapped group excluded
   tg.downstream_task_ids  {'after'} -> set()
   ```
   
   **3. Breadth on the clear endpoint.** Counting only legal `POST /dags/{dag_id}/clearTaskInstances` requests that should answer 200:
   
   | Dag shape | before | with this PR |
   |---|---|---|
   | mapped group holding a nested group | 14 of 20 fail | 0 of 20 |
   | mapped group holding a mapped group | 8 of 12 fail | 0 of 12 |
   | plain group nested in a loop body | 8 of 16 fail | 0 of 16 |
   
   All 30 are the same `ReferenceError`, and five of them pass neither include flag. This one line takes all three shapes to zero and restores the edges in point 2.
   
   Two more call sites worth listing alongside the ones in the issue: `cli/commands/task_command.py`, and `SerializedDAG.set_task_instance_state`, `set_task_group_state` and `clear`.
   
   Happy to contribute the reproduction as a test here if that is useful.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org