Serialize mapped DagParam as dag_param instead of a memory address by Vamsi-klu · Pull Request #72242 · apache/airflow

Vamsi-klu · 2026-08-29T02:50:05Z

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?

Yes — Grok 4.6

Generated-by: Grok 4.6 following the guidelines

Drafted-by: Grok 4.6 (no human review before posting)

DagParam in mapped .partial() fell through serialize() to str(), which embeds a process-local address. Every parse then wrote a new Dag version even though the Dag file did not change.

@task

Only reconstruct SerializedDagParam from {__type: dag_param} template fields. Cover resolve() when conf omits the name, when nothing is resolvable, and address-stable nested @task calls.

potiuk · 2026-08-29T13:22:09Z

CC: @ephraimbuddy @jedcunningham I know you are working on general DagVersion stabilisation - for 3.3.2 - maybe you can have a look here - this one seems to have a merit.

potiuk

Reviewed with the patch applied locally on the PR's base (82215cf). The core fix is right and well-targeted: the root cause matches the issue reporter's own diagnosis, and the DAT.DAG_PARAM pair mirrors the existing XComArg treatment.

What I ran:

Check	Result
11 new tests	pass
full `airflow-core/tests/unit/serialization/` (588 tests)	pass
`ruff check` / `ruff format --check`	clean
`prek run mypy-airflow-core`	pass
revert `serialized_objects.py` only	6 of 11 new tests fail - confirms the experiment in the PR body

Two comments below suggest cutting scope. If both are taken, the PR shrinks to the enums.py line, the two serialize()/deserialize() branches, a data-only SerializedDagParam, and ~6 tests - which is exactly what the issue asks for.

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

potiuk · 2026-08-29T18:39:19Z

            return cls._encode(cls._serialize_param(var), type_=DAT.PARAM)
        elif isinstance(var, XComArg):
            return cls._encode(serialize_xcom_arg(var), type_=DAT.XCOM_REF)
+        elif isinstance(var, (DagParam, SerializedDagParam)):


This is the actual fix and it looks right. Serializing default recursively rather than passing it through is the detail that matters: NOTSET round-trips as arg_not_set instead of the literal string "NOTSET". Identity holds too - task-sdk/.../_internal/types.py:38 re-exports core's NOTSET singleton when core is installed, so the is not NOTSET check compares the same object deserialize() returns.

potiuk · 2026-08-29T18:39:19Z

                value = getattr(op, template_field, None)
                if not cls._is_excluded(value, template_field, op):
-                    serialize_op[template_field] = serialize_template_field(value, template_field)
+                    if isinstance(value, (DagParam, SerializedDagParam)):


Suggest dropping this hunk (and its pair in populate_operator). The issue is about partial_kwargs; template fields don't have the version-inflation bug.

On main today a directly-assigned DagParam already serializes stably, because serialize_template_field reaches DagParam.serialize through its callable(inspect.getattr_static(obj, "serialize", None)) branch (helpers.py:80). I checked:

main: MockOperator(task_id="t", arg1=dag.param("p","d")) -> "arg1": {"dag_id": "...", "default": "d", "name": "p"} # no memory address

So there's no stability gain here - but there is a cost. With the hunk, the deserialized scheduler-side attribute is an object rather than JSON:

json.dumps(restored.task_dict["t"].arg1) # TypeError: Object of type SerializedDagParam is not JSON serializable

Making template fields JSON-encodable is serialize_template_field's stated first responsibility (helpers.py:43). I traced the consumers and did not find a live break - get_serialized_template_fields() (renderedtifields.py:85) takes a SerializedBaseOperator and re-enters serialize_template_field, which handles the new object through that same .serialize() branch. So this isn't a bug today; it just widens the blast radius of a bugfix PR.

It also only tags a directly assigned DagParam, leaving an asymmetry:

arg1=dag.param("p","d") -> {"__type": "dag_param", ...} -> SerializedDagParam arg1=[dag.param("p","d")] -> [{"dag_id": ..., ...}] -> plain dict arg1={"k": dag.param(...)} -> {"k": {...}} -> plain dict

If template fields should be tagged, it needs to recurse uniformly and the encodability contract needs addressing explicitly - worth its own PR either way.

Minor: SerializedDagParam in this isinstance tuple is unreachable - _serialize_node takes an SdkOperator.

potiuk · 2026-08-29T18:39:19Z

            # Use centralized field deserialization logic
            if k in encoded_op.get("template_fields", []):
-                pass  # Template fields are handled separately
+                if isinstance(v, dict) and v.get(Encoding.TYPE) == DAT.DAG_PARAM and Encoding.VAR in v:


Pair of the comment above - suggest dropping this too. It also only unwraps a DagParam at the top level of a template field, so arg1=[dag.param(...)] stays a plain dict while arg1=dag.param(...) becomes a SerializedDagParam.

potiuk · 2026-08-29T18:39:19Z

+        self.name = name
+        self.default = default
+
+    def iter_references(self):


Both of these methods are unreachable in airflow-core; suggest dropping them and keeping SerializedDagParam as a plain data holder.

resolve(): nothing resolves scheduler-side objects - the worker re-parses the Dag file with the SDK, so the real DagParam.resolve runs there. The precedent is right next door: the whole SchedulerXComArg family in definitions/xcom_arg.py deliberately has no resolve(), only iter_references() and map-length helpers. I also confirmed scheduler-side partial_kwargs is read only for scheduling attributes (owner, retries, pool, ...) in definitions/mappedoperator.py:146-262; user kwargs are never touched.

iter_references(): SchedulerXComArg.iter_xcom_references (xcom_arg.py:74) only dispatches to arg.iter_references() for ReferenceMixin instances, and SerializedDagParam isn't one.

As written it's ~12 lines hand-copied from DagParam.resolve with nothing keeping the two in sync, and the three test_serialized_dagparam_resolve_* tests exercise only themselves. If the resolve semantics are deliberately kept for something planned, a comment saying so would help - otherwise the next reader will reasonably assume the scheduler resolves params.

Also: iter_references is missing a return annotation, unlike the rest of the file.

potiuk · 2026-08-29T18:39:19Z

    TASK_GROUP = "taskgroup"
    EDGE_INFO = "edgeinfo"
    PARAM = "param"
+    DAG_PARAM = "dag_param"


Non-blocking, just so it's a conscious call: a new __type value means an older Airflow reading a newly written blob hits TypeError: Invalid type dag_param in deserialization, and SERIALIZER_VERSION stays at 3. Consistent with how other DAT members were added, and Airflow expects components on one version - flagging only because rolling upgrades touch this path.

potiuk · 2026-08-29T18:39:19Z

+        param.resolve({"dag_run": type("DR", (), {"conf": {}})(), "params": {}})
+
+
+def test_dagparam_nested_in_taskflow_call_is_address_stable():


This test passes on unmodified main - I ran it. do(dag.param(...)) is a non-mapped taskflow task, so op_kwargs goes through serialize_template_field, which was already address-stable. It pins pre-existing behaviour rather than this change.

Per AGENTS.md ("every test must fail without the PR's change") this should either go or be reshaped into something that does fail without the fix - e.g. asserting the dag_param type tag.

Same applies more mildly to test_dagparam_jinja_string_in_partial_stays_string (already noted in the PR body); that one reads as a deliberate negative guard, so it's more defensible.

ephraimbuddy

Approach is right — a dag_param DAT tag mirroring XComArg is what the issue asked for, and I verified the fix works: on main the mapped op_kwargs serializes to "<airflow.sdk.definitions.param.DagParam object at 0x...>", with this PR it is a stable typed encoding. I also confirmed .expand(x=dag.param(...)) really is rejected at authoring, so encode_expand_input is not a gap. Full airflow-core/tests/unit/serialization/ passes (602); the one failure and one error I saw reproduce identically on the base commit and are environmental.

Two things I would like trimmed before merge: a resolve() that nothing calls, and a template-field change that is a behaviour change rather than part of the fix. Details inline.

Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting

ephraimbuddy · 2026-08-29T18:40:27Z

+    def iter_references(self):
+        return ()
+
+    def resolve(self, context: Mapping[str, Any]) -> Any:


resolve() is unreachable. Nothing on the scheduler or API side resolves values out of a deserialized Dag — runtime resolution happens in the Task SDK against the real DagParam, because workers parse the Dag file themselves. The deliberate precedent is right next door: SchedulerXComArg in serialization/definitions/xcom_arg.py has no resolve() at all, for exactly this reason.

iter_references() is dead too — SchedulerXComArg.iter_xcom_references dispatches on ReferenceMixin, which SerializedDagParam does not subclass, so it is never reached.

Could we drop both methods along with the three test_serialized_dagparam_resolve_* tests? A hand-copied mirror of DagParam.resolve that nothing calls will drift from the SDK silently, and the tests only prove the copy matches itself.

Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting

ephraimbuddy · 2026-08-29T18:40:27Z

        }


+class SerializedDagParam:


Minor: no __eq__, so two structurally identical params compare unequal. The siblings in this package use attrs.define, which would give you __eq__ and __repr__ for free.

Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting

ephraimbuddy · 2026-08-29T18:40:27Z

                value = getattr(op, template_field, None)
                if not cls._is_excluded(value, template_field, op):
-                    serialize_op[template_field] = serialize_template_field(value, template_field)
+                    if isinstance(value, (DagParam, SerializedDagParam)):


This branch is a behaviour change, not part of the fix. On main a non-mapped bash_command=dag.param("cmd", "echo hi") already serializes stably, because serialize_template_field picks up the serialize() method:

main: "bash_command": {"dag_id": "probe", "default": "echo hi", "name": "cmd"} PR: "bash_command": {"__var": {...}, "__type": "dag_param"}

There was never a memory address on this path. Changing it rewrites the blob for every existing Dag that uses dag.param() in a template field — one new DagVersion on upgrade, which is the thing this PR exists to prevent — and flips the deserialized value from dict to SerializedDagParam.

The typed form is arguably better, so I am not asking to revert it outright; I am asking whether it is intended. If it stays, please say so in the PR description and widen the newsfragment, which currently only mentions mapped .partial().

Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting

ephraimbuddy · 2026-08-29T18:40:27Z

            # Use centralized field deserialization logic
            if k in encoded_op.get("template_fields", []):
-                pass  # Template fields are handled separately
+                if isinstance(v, dict) and v.get(Encoding.TYPE) == DAT.DAG_PARAM and Encoding.VAR in v:


Only a top-level DagParam in a template field gets the typed encoding, so nesting behaves differently depending on whether the task is mapped:

BashOperator(task_id="nested", env={"A": dag.param("e", "ev")}) BashOperator.partial(task_id="mnested", env={"A": dag.param("m", "mv")}).expand(...)

restored non-mapped env: {'A': {'dag_id': 'p2', 'default': 'ev', 'name': 'e'}} restored mapped env : {'A': SerializedDagParam(dag_id='p2', name='m')}

Same author-level construct, two deserialized types. Neither is unstable, so this is not a correctness bug today, but it is the kind of asymmetry that bites later.

Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting

ephraimbuddy · 2026-08-29T18:40:27Z

            return cls._encode(cls._serialize_param(var), type_=DAT.PARAM)
        elif isinstance(var, XComArg):
            return cls._encode(serialize_xcom_arg(var), type_=DAT.XCOM_REF)
+        elif isinstance(var, (DagParam, SerializedDagParam)):


Is there a path that re-serializes an already-deserialized Dag? I could not find one, which would make the SerializedDagParam half of this isinstance dead. Happy to be wrong if you know of a caller.

Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting

ephraimbuddy · 2026-08-29T18:40:27Z

+        param.resolve({"dag_run": type("DR", (), {"conf": {}})(), "params": {}})
+
+
+def test_dagparam_nested_in_taskflow_call_is_address_stable():


This one passes on main — do(dag.param(...)) on a non-mapped @task routes through serialize_template_field, so that line of the issue's repro was never broken. Fine to keep as a regression guard, but the PR description's "6 failed for the right reason" overstates what is actually pinned; test_dagparam_in_non_mapped_operator_field is similar, in that its "object at 0x" assertion is already true on main and only the isinstance(..., SerializedDagParam) assertion is new.

Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting

ephraimbuddy · 2026-08-29T18:40:27Z

            OperatorSerialization.serialize(op)
+
+
+def _encoded_dag_params(obj):


Both fallbacks here are redundant: Encoding.TYPE is "__type" and DagAttributeTypes.DAG_PARAM == "dag_param", both being str enums. obj.get(Encoding.TYPE) == DagAttributeTypes.DAG_PARAM is enough.

Drafted-by: Claude Code (Opus 5); reviewed by @ephraimbuddy before posting

Stop mapped Dag version churn from DagParam memory addresses

47740c4

DagParam in mapped .partial() fell through serialize() to str(), which embeds a process-local address. Every parse then wrote a new Dag version even though the Dag file did not change.

boring-cyborg Bot added the area:DAG-processing label Aug 29, 2026

Add newsfragment for stable DagParam serialization

3534acd

Vamsi-klu marked this pull request as ready for review August 29, 2026 03:32

Vamsi-klu requested review from ashb and bolkedebruin as code owners August 29, 2026 03:32

Vamsi-klu and others added 2 commits August 29, 2026 03:57

Limit template-field deserialize to dag_param encodings

939b3cc

Only reconstruct SerializedDagParam from {__type: dag_param} template fields. Cover resolve() when conf omits the name, when nothing is resolvable, and address-stable nested @task calls.

Merge branch 'main' into fix/68941-dagparam-partial-serialization

62fd168

potiuk reviewed Aug 29, 2026

View reviewed changes

ephraimbuddy reviewed Aug 29, 2026

View reviewed changes

		param.resolve({"dag_run": type("DR", (), {"conf": {}})(), "params": {}})


		def test_dagparam_nested_in_taskflow_call_is_address_stable():

		OperatorSerialization.serialize(op)


		def _encoded_dag_params(obj):

		}


		class SerializedDagParam:

Uh oh!

Conversation

Vamsi-klu commented Aug 29, 2026

What is the change?

Why did I do it?

How did I do it?

What's the impact?

What's the test plan?

Was generative AI tooling used to co-author this PR?

Uh oh!

potiuk commented Aug 29, 2026

Uh oh!

potiuk left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ephraimbuddy left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants