You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Desdroid (via GitHub)" <gi...@apache.org> on 2026/06/23 13:07:07 UTC
[PR] [Serialization] Fix: Serialize DagParam like a normal Param. [airflow]
Desdroid opened a new pull request, #68901:
URL: https://github.com/apache/airflow/pull/68901
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
When we use a DagParam as a kwarg to `partial()` of a dynamically mapped task, the DagParam will be serialized with its memory address. This leads to a DAG version inflation, as the memory address is not stable.
This is a simple repro:
```python
from __future__ import annotations
import json
from airflow.sdk import DAG
from airflow.sdk import task
from airflow.serialization.definitions.param import SerializedParam
from airflow.serialization.serialized_objects import DagSerialization
if __name__ == "__main__":
with DAG(dag_id="repro_dagparam") as dag:
@task
def add(value):
return value
@task
def do(something):
return something
do(dag.param("some", "some_default_val"))
add.partial(value=dag.param("p", "p_default_val")).expand(value=[1, 2, 3])
ser = DagSerialization.to_dict(dag)
mapped = ser["dag"]["tasks"][1]["__var"]
print(json.dumps(mapped.get("partial_kwargs"), indent=2, default=str))
print("ROUNDTRIP:")
deser = DagSerialization.from_dict(ser)
t = deser.get_task("add")
print(type(t.partial_kwargs.get("op_kwargs")), t.partial_kwargs.get("op_kwargs"))
```
This PR fixes this issue by serializing the DagParam as a normal Param. This behavior matches the serialization that using the `DagParam` as a normal arg does (like in the `do` task - it ends up being serialized as a `Param`).
__Note: I am not 100% sure that this is the correct fix as I'm not super deep in the DAG serialization code, but it seems to work.__
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#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](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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] [Serialization] Fix: Serialize DagParam like a normal Param. [airflow]
Posted by "Desdroid (via GitHub)" <gi...@apache.org>.
Desdroid commented on PR #68901:
URL: https://github.com/apache/airflow/pull/68901#issuecomment-4788603217
Converted to Issue #68941
--
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] [Serialization] Fix: Serialize DagParam like a normal Param. [airflow]
Posted by "Desdroid (via GitHub)" <gi...@apache.org>.
Desdroid commented on code in PR #68901:
URL: https://github.com/apache/airflow/pull/68901#discussion_r3466389347
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -580,6 +579,10 @@ def serialize(
return TaskGroupSerialization.serialize_task_group(var)
elif isinstance(var, Param):
return cls._encode(cls._serialize_param(var), type_=DAT.PARAM)
+ elif isinstance(var, DagParam):
+ return cls._encode(
+ cls._serialize_param(Param(default=var._default, source="dag")), type_=DAT.PARAM
+ )
Review Comment:
You're right, I don't think that's the correct thing to do. In `op_args` the serialized representation is used after DagParam went through `serialize_template_field`. For the `partial_kwargs` this path is not taken and what I did here will not work properly on runtime, i got confused while debugging in the heat.
I don't really know how a fix for this should look like - I think either `serialize_mapped_operator` needs to be adapted and possibly should use `serialize_template_field` for the `partial_kwargs` ops or an own Serialization class is needed for `DagParam` possibly similar to `XComArg`?
If somebody is willing to point me in a direction, I am also open to contribute a fix.
I'm closing this PR and will open a bug report instead.
We found a workaround using Jinja Templates for DagParams in partial for now.
```pyhton
add.partial(value="{{ params.p }}").expand(value=[1, 2, 3])
```
Sorry for the confusion.
--
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] [Serialization] Fix: Serialize DagParam like a normal Param. [airflow]
Posted by "yuseok89 (via GitHub)" <gi...@apache.org>.
yuseok89 commented on code in PR #68901:
URL: https://github.com/apache/airflow/pull/68901#discussion_r3465807563
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -580,6 +579,10 @@ def serialize(
return TaskGroupSerialization.serialize_task_group(var)
elif isinstance(var, Param):
return cls._encode(cls._serialize_param(var), type_=DAT.PARAM)
+ elif isinstance(var, DagParam):
+ return cls._encode(
+ cls._serialize_param(Param(default=var._default, source="dag")), type_=DAT.PARAM
+ )
Review Comment:
Have you tested this? Looking at the code, it seems only _default is stored and _name is dropped, and since DagParam resolves at runtime via dag_run.conf[self._name], I think this could be a problem. Or am I misreading the flow?
--
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] [Serialization] Fix: Serialize DagParam like a normal Param. [airflow]
Posted by "Desdroid (via GitHub)" <gi...@apache.org>.
Desdroid closed pull request #68901: [Serialization] Fix: Serialize DagParam like a normal Param.
URL: https://github.com/apache/airflow/pull/68901
--
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] [Serialization] Fix: Serialize DagParam like a normal Param. [airflow]
Posted by "Desdroid (via GitHub)" <gi...@apache.org>.
Desdroid commented on PR #68901:
URL: https://github.com/apache/airflow/pull/68901#issuecomment-4780340376
@ephraimbuddy @wjddn279 Since you were working on serialization in #63871 - what do you think of this fix I propose here? Does this make sense or is it wrong and some change is needed in `serialize_template_field`?
The DagParam gets serialized with it's memory address when in `partial` of DTM.
--
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