Vamsi-klu opened a new pull request, #72242:
URL: https://github.com/apache/airflow/pull/72242
## What is the change?
`DagParam` in mapped `.partial()` (and in ordinary operator fields) now
serializes as `{__type: dag_param, __var: {dag_id, name, default}}` instead of
`str(DagParam)`, which included a process-local memory address. Deserialize
rebuilds a `SerializedDagParam` that resolves like SDK `DagParam`.
## Why did I do it?
closes: #68941
`BaseSerialization.serialize()` handled `Param` and `XComArg`, then fell
through to `str(var)`. That string changes every parse, so `DagVersion`
inflated even when the Dag file did not. Jinja `"{{ params.p }}"` was a
workaround, not a fix; `dag.param(...)` is the documented API.
A previous attempt (#69091) closed as a stale draft. This PR does not
reconstruct a live SDK `DagParam` during DAG hydration. `resolve()` does not
need `current_dag`.
## How did I do it?
Added `DAT.DAG_PARAM` and treated `DagParam` like `XComArg` in `serialize()`
/ `deserialize()`. The default is itself serialized so `NOTSET` stays
`arg_not_set`, not the string `"NOTSET"`. Template fields used
`serialize_template_field`, which called `DagParam.serialize()` and dropped the
type tag, so those fields now go through `serialize()` as well. I did not reuse
`DAT.PARAM`; that type is a schema-plus-default object, not a late-bound name.
`MockOperator.expand(arg1=dag.param(...))` is rejected at authoring
(`unexpected type 'DagParam'`), so `encode_expand_input` never sees a
`DagParam` today.
## What's the impact?
Mapped Dags that pass `dag.param(...)` through `.partial()` stop writing a
new Dag version on every parse. Old serialized rows that already contain
`object at 0x...` are left as-is; the next successful parse writes a stable
blob.
## What's the test plan?
New tests in `test_dag_serialization.py`:
- `@task` + `.partial(value=dag.param(...)).expand(...)`: no `object at 0x`,
`__type` is `dag_param`, two independently built Dags serialize equal
- roundtrip `from_dict` yields `SerializedDagParam` with the right
name/default
- classic mapped operator `.partial(arg1=dag.param(...))`
- non-mapped operator field
- `NOTSET` default is not stringified; `strict=True` encodes instead of
raising
- Jinja `"{{ params.p }}"` in `.partial()` stays a string
- two DagParams in one `.partial()`
- `SerializedDagParam.resolve` prefers `dag_run.conf`
I reverted the serialize/deserialize branches and re-ran those tests: 6
failed for the right reason (`object at 0x` / `SerializationError`). The Jinja
test and the direct `resolve()` test still passed, as expected.
```
uv run --project airflow-core pytest \
airflow-core/tests/unit/serialization/test_dag_serialization.py \
-k 'dagparam or DagParam or SerializedDagParam or operator_expand_xcomarg
or taskflow_expand_serde or dag_params_roundtrip'
```
13 passed. Ruff and airflow-core mypy passed via prek.
`check-schema-defaults` was skipped locally because it requires a CI image; the
change does not touch operator/DAG schema defaults. `schema.json` was not
edited; `DagSerialization.to_dict` already validates and accepted `dag_param`.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Grok 4.6
Generated-by: Grok 4.6 following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
Drafted-by: Grok 4.6 (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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
