[Serialization] Fix: Serialize DagParam like a normal Param.

[Desdroid]

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:

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?

• [] Yes (please specify the tool below)

---

• Read the 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) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[Desdroid]

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

[yuseok89]

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?

[Desdroid]

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.

add.partial(value="{{ params.p }}").expand(value=[1, 2, 3])

Sorry for the confusion.

[Desdroid]

Converted to Issue #68941