[PR] Make template_fields_extra iteration order stable across processes [airflow]

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

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

   <!--
   Please provide above a brief description of the changes made in this pull request.
   -->
   
   `BaseOperator.template_fields_extra` is currently typed `ClassVar[frozenset[str]] = frozenset()`. The PR that introduced it (apache/airflow#74920) iterates that value in three places to build the merged `template_fields` tuple and the `MappedOperator.template_fields_extra` value. **`frozenset` iteration order is not stable across Python processes** — string hashing in CPython 3.3+ is randomized per process via `PYTHONHASHSEED`. The result is a different `template_fields_extra` / merged `template_fields` order in the Dag processor, scheduler, worker, and tests, and a different serialized value in the metadata DB on every parse.
   
   This change is stacked on top of apache/airflow#74920, so the two PRs land together. The diff is intentionally small: re-type the attribute as `ClassVar[Sequence[str]] = ()`, document that `tuple` (or another ordered sequence) is the supported form, harden the three materialisation sites, fix the order/dedup tests so they assert against `tuple` rather than `frozenset`, and add one regression test that spawns a subprocess with a fresh `PYTHONHASHSEED` to prove the order is process-stable.
   
   ### Why
   
   ```text
   $ for i in 1 2 3 4 5; do python3 -c "print(tuple(frozenset({'alpha','beta','gamma'})))"; done
   ('gamma', 'beta', 'alpha')
   ('beta', 'alpha', 'gamma')
   ('alpha', 'gamma', 'beta')
   ('gamma', 'alpha', 'beta')
   ('beta', 'gamma', 'alpha')
   ```
   
   Concrete failure modes that this fixes:
   
   - `test_template_fields_extra_preserves_order_and_deduplicates` and `test_template_fields_extra_preserves_order_and_deduplicates_through_mapped_operator` in `task-sdk/tests/task_sdk/bases/test_operator.py` both asserted a deterministic order from a `frozenset` — that order only held in processes whose `PYTHONHASHSEED` happened to iterate the set that way. Running under a different seed flipped the list to `["arg1", "arg3", "arg2"]` and the assertion failed.
   - The serialization tests in `airflow-core/tests/unit/serialization/test_dag_serialization.py` already worked around the instability: `test_serialize_template_fields_extra_roundtrip` and `test_serialize_template_fields_extra_roundtrip_mapped` used `sorted()` / `set()` on the result. The author of #74920 felt the instability and patched the test instead of fixing the data type.
   - The per-process order flowed into `SerializedBaseOperator.template_fields_extra` (written to the metadata DB) and into the `dag_version_diff` field list, so the same Dag could yield a different serialized version on every reparse — the same class of bug as the recently-closed apache/airflow#53957.
   
   ### What changes
   
   - `task-sdk/src/airflow/sdk/bases/operator.py` — `BaseOperator.template_fields_extra: ClassVar[Sequence[str]] = ()`. The `__init__` merge still uses `dict.fromkeys((*self.template_fields, *tuple(extra)))` so declaration order is preserved, and an inline comment calls out why `Sequence` (and not `frozenset`) is required.
   - `task-sdk/src/airflow/sdk/definitions/mappedoperator.py` — `OperatorPartial.__call__` materialises the merged `template_fields` and the per-instance `template_fields_extra` from a `tuple` of the class attribute, so the result is order-stable. The `MappedOperator.template_fields_extra` type is already `Sequence[str]`.
   - `task-sdk/src/airflow/sdk/bases/decorator.py` — the `@task` decorator factory takes the same shape.
   - `airflow-core/src/airflow/serialization/definitions/{baseoperator,mappedoperator}.py` — `template_fields_extra` is tightened from `Collection[str]` to `Sequence[str]` so the order contract is explicit in the type.
   - `airflow-core/docs/howto/custom-operator.rst` — the `template_fields_extra` example is updated to `tuple[str, ...] = ("world",)` and a note explains why `frozenset` and `set` are not supported.
   - `task-sdk/tests/task_sdk/bases/test_operator.py` — the four `template_fields_extra` declarations that used `frozenset({...})` are now `tuple` literals; the order/dedup tests assert against the declared order verbatim; the empty-default test now also asserts that the default is a `Sequence` (a `tuple`), not a `frozenset`. A new `test_template_fields_extra_order_is_stable_across_processes` spawns three subprocesses with a fresh `PYTHONHASHSEED` and asserts they all produce the same `template_fields` order.
   - `airflow-core/tests/unit/serialization/test_dag_serialization.py` — the two `template_fields_extra` round-trip tests use `tuple` declarations and assert `list(...) == [...]` (order preserved) instead of `set(...)` / `sorted(...)` (order discarded). The expected `template_fields_extra: ()` value in the `BaseOperator` fixture is updated to `()`.
   
   ### Tests
   
   ```text
   $ .venv/bin/pytest task-sdk/tests/task_sdk/bases/test_operator.py -k template_fields_extra -v
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_class_level_merges_into_template_fields PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_renders_extra_string_fields PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_creates_xcomarg_upstream_dependency PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_default_is_empty_sequence PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_preserves_order_and_deduplicates[extras-partly-overlap-template_fields] PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_preserves_order_and_deduplicates[extras-fully-overlap-template_fields] PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_preserves_order_and_deduplicates[extras-and-template_fields-reverse-order] PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_preserves_order_and_deduplicates[extras-empty] PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_preserves_order_and_deduplicates_through_mapped_operator PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_propagates_through_mapped_operator PASSED
   task-sdk/tests/task_sdk/bases/test_operator.py::TestBaseOperator::test_template_fields_extra_order_is_stable_across_processes PASSED
   ================ 11 passed, 122 deselected, 1 warning in 2.47s =================
   ```
   
   Serialization round-trip was verified end-to-end with a small driver (since `test_dag_serialization.py` requires the `cncf.kubernetes` provider for unrelated reasons and would not load in this sandbox); the equivalent assertions are now part of `test_serialize_template_fields_extra_roundtrip` and `test_serialize_template_fields_extra_roundtrip_mapped` and assert `list(...) == [...]` instead of `set(...)` / `sorted(...)`.
   
   `prek run ruff` and `prek run ruff-format` pass on every touched file.
   
   closes: #74991
   
   ---
   
   ##### Was generative AI tooling used to co-author this PR?
   
   - [X] Yes
   
   <!-- Generated-by: opencode (minimax/MiniMax-M3) following https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst -->
   
   ---
   Drafted-by: opencode (minimax/MiniMax-M3) (no human review before posting)
   


-- 
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