This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 37ab48778c9 Preserve explicit falsy partial() values on mapped
operators (#75679)
37ab48778c9 is described below
commit 37ab48778c9cce8abca6ff6b5e0a61234c1377aa
Author: Aboulfoul <[email protected]>
AuthorDate: Sat Nov 7 08:26:34 2026 +0100
Preserve explicit falsy partial() values on mapped operators (#75679)
When [core] default_task_retries is non-zero,
Op.partial(retries=0).expand(...)
serialized without retries, so the scheduler filled it from client_defaults
and the task retried default_task_retries times. The explicit-override check
added in #57853 looked at DAG default_args and constructor kwargs, but
mapped
operators keep explicit values in partial_kwargs.
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../airflow/serialization/serialized_objects.py | 4 ++++
.../unit/serialization/test_dag_serialization.py | 25 ++++++++++++++++++++++
2 files changed, 29 insertions(+)
diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py
b/airflow-core/src/airflow/serialization/serialized_objects.py
index 782d4bd35e0..93ab896c3c3 100644
--- a/airflow-core/src/airflow/serialization/serialized_objects.py
+++ b/airflow-core/src/airflow/serialization/serialized_objects.py
@@ -1496,6 +1496,10 @@ class OperatorSerialization(DAGNode, BaseSerialization):
and op._BaseOperator__init_kwargs[attrname] == var
):
return False
+ # Mapped operators carry explicit values in
partial_kwargs, not __init_kwargs.
+ partial_kwargs = getattr(op, "partial_kwargs", None)
+ if partial_kwargs and attrname in partial_kwargs and
partial_kwargs[attrname] == var:
+ return False
# If client_defaults doesn't have this field (matches schema
default),
# exclude for optimization even if in default_args
diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py
b/airflow-core/tests/unit/serialization/test_dag_serialization.py
index 4bc57bed583..34b4376796a 100644
--- a/airflow-core/tests/unit/serialization/test_dag_serialization.py
+++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py
@@ -4984,6 +4984,31 @@ class TestMappedOperatorSerializationAndClientDefaults:
# Explicit values should override client_defaults
assert deserialized_task.retries == 5 # Explicitly set value
+ @pytest.mark.parametrize(
+ ("default_args", "partial_retries", "expected_retries"),
+ [
+ pytest.param({}, 0, 0, id="partial-zero"),
+ pytest.param({"retries": 1}, 0, 0,
id="partial-zero-overrides-default-args"),
+ pytest.param({}, None, 3, id="unset-uses-client-default"),
+ ],
+ )
+ def test_mapped_operator_preserves_explicit_schema_default(
+ self, default_args, partial_retries, expected_retries,
operator_defaults
+ ):
+ """An explicit partial() value equal to the schema default must
survive when client_defaults differ."""
+ partial_kwargs = {} if partial_retries is None else {"retries":
partial_retries}
+ with operator_defaults({"retries": 3}):
+ with DAG(dag_id="test_mapped_explicit_schema_default",
default_args=default_args) as dag:
+ BashOperator.partial(task_id="mapped_task",
**partial_kwargs).expand(
+ bash_command=["echo 1", "echo 2"]
+ )
+
+ serialized = DagSerialization.to_dict(dag)
+ assert serialized["client_defaults"]["tasks"]["retries"] == 3
+
+ deserialized_task =
DagSerialization.from_dict(serialized).get_task("mapped_task")
+ assert deserialized_task.retries == expected_retries
+
@pytest.mark.parametrize(
("task_config", "dag_id", "task_id", "non_default_fields"),
[
