apache / apache/airflow

Dag Version Inflation: DagParam serialized with memory address if used in partial of a mapped task

Open
#68,941 2 comments 0 reactions 0 assignees View on GitHub
area:core area:dynamic-task-mapping area:serialization kind:bug
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

### Under which category would you file this issue?

Airflow Core

### Apache Airflow version

3.2.2

### What happened and how to reproduce it?

When using a DagParam as a kwarg to `partial` it will get serialized with its memory address. As the address is not stable this leads to version inflation.

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 will print
```
{
"op_kwargs": {
"__var": {
"value": ""
},
"__type": "dict"
},
"op_args": []
}
```
Note the memory address ``

### What you think should happen instead?

_No response_

### Operating System

_No response_

### Deployment

Official Apache Airflow Helm Chart

### Apache Airflow Provider(s)

_No response_

### Versions of Apache Airflow Providers

_No response_

### Official Helm Chart version

1.22.0 (latest released)

### Kubernetes Version

_No response_

### Helm Chart configuration

_No response_

### Docker Image customizations

_No response_

### Anything else?

A workaround exists by using Jinja templates to pass DagParams in partial
```python
add.partial(value="{{ params.p }}").expand(value=[1, 2, 3])
```

The root cause is in [`serialize()`](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/serialization/serialized_objects.py#L457) not having a case for `DagParam` running into the default branch and serializing the object as string.
I think either a separate `DagParam` case needs to be added to [`serialize()`](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/serialization/serialized_objects.py#L457) similar to `XComArg` or [`serialize_mapped_operator()`](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/serialization/serialized_objects.py#L963) needs to handle the `partial_kwargs` ops with [`serialize_template_field()`](https://github.com/apache/airflow/blob/main/airflow-core/src/airflow/serialization/helpers.py#L37) or something similar.
If somebody can point me in a direction for a fix, I am willing to contribute a PR.

### Are you willing to submit PR?

- [x] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Contributor guide

Open the contributing guide

Research direction

Run the supplied reproduction first, then read serialize() and serialize_mapped_operator() in airflow-core/src/airflow/serialization/serialized_objects.py, along with serialize_template_field() in airflow-core/src/airflow/serialization/helpers.py. Compare the serialized partial_kwargs and the DagSerialization round trip; done means a DagParam no longer becomes a memory-address string and the serialized representation remains stable.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
68/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.