jason810496 commented on code in PR #75267:
URL: https://github.com/apache/airflow/pull/75267#discussion_r4334550359
##########
go-sdk/airflow/serialize.go:
##########
@@ -202,6 +202,8 @@ type specRules struct {
// set names the lists that Python keeps in a set, so that Python
writes the list sorted and
// without repeats. writeSpecFields writes it the same way.
set map[string]bool
+ // keepDefaults preserves an explicitly set task field when it equals
the schema default.
+ keepDefaults bool
Review Comment:
A per-SDK flag for a rule that should hold everywhere. Dropping it here, in
`taskSpecRules` and in the write condition leaves `schemaDefault` and
`isSameJSON` with no reader.
```suggestion
```
##########
go-sdk/airflow/serialize.go:
##########
@@ -218,7 +220,8 @@ var dagSpecRules = specRules{
}
var taskSpecRules = specRules{
- fields: taskSpecFields,
+ fields: taskSpecFields,
+ keepDefaults: true,
Review Comment:
```suggestion
fields: taskSpecFields,
```
##########
go-sdk/airflow/serialize.go:
##########
@@ -252,7 +256,7 @@ func writeSpecFields(data map[string]any, spec any, rules
specRules) {
continue
}
encoded, set := encodeSpecValue(value.Field(i), rules.set[name])
- if !set || (field.schemaDefault != nil && isSameJSON(encoded,
field.schemaDefault)) {
+ if !set || (!rules.keepDefaults && field.schemaDefault != nil
&& isSameJSON(encoded, field.schemaDefault)) {
Review Comment:
```suggestion
if !set {
```
##########
go-sdk/airflow/serialize.go:
##########
@@ -234,10 +237,11 @@ var taskSpecRules = specRules{
// from the defaults of Python's TaskGroup instead.
var taskGroupSpecRules = specRules{fields: taskGroupSpecFields}
-// writeSpecFields writes into data each field of spec that rules does not
skip. It leaves out a
-// field that is unset or that holds its schema default. A field is unset when
it holds its zero
-// value. A pointer field is unset only when it is nil, so a pointer field can
set the zero value of
-// the type it points to, such as false or 0.
+// writeSpecFields writes into data each field of spec that rules does not
skip. It leaves out
+// fields that are unset, and normally those that hold their schema default.
Task rules preserve
+// explicitly set schema-default values so Airflow config defaults cannot
replace them. A field
+// is unset when it holds its zero value. A pointer field is unset only when
nil, so it can explicitly
+// set the zero value of its type, such as false or 0.
Review Comment:
```suggestion
// writeSpecFields writes into data each field of spec that rules does not
skip. It leaves out only
// the fields that are unset: Airflow fills an unset config-backed field
from its own config, so a
// value the author set has to be written even when it equals the schema
default. A field is unset
// when it holds its zero value. A pointer field is unset only when nil, so
it can explicitly set
// the zero value of its type, such as false or 0.
```
##########
ts-sdk/src/coordinator/serde.ts:
##########
@@ -429,8 +433,9 @@ function simpleTimetable(type: string): SerializedValue {
}
/**
- * Write the fields a spec set onto `data`, skipping any left at its schema
- * default — Python's serializer omits what the scheduler re-derives.
+ * Write the fields a spec set onto `data`. By default, omit schema-default
+ * values that Python's serializer lets the scheduler re-derive; task rules can
+ * preserve them when the author explicitly set them.
Review Comment:
```suggestion
* Write every field a spec set onto `data`. A field left unset stays out so
* Airflow can fill the config-backed ones from its own config; a value the
* author set is written even when it equals the schema default.
```
##########
ts-sdk/src/coordinator/serde.ts:
##########
@@ -142,6 +144,8 @@ const DAG_FIELD_RULES: FieldRules = {
const TASK_FIELD_RULES: FieldRules = {
fields: TASK_SCHEMA_FIELDS,
decorated: new Set(["executor_config"]),
+ // Preserve explicit values such as retries=0 so Airflow config defaults do
not replace them.
+ keepDefaults: true,
Review Comment:
```suggestion
```
##########
ts-sdk/src/coordinator/serde.ts:
##########
@@ -131,6 +131,8 @@ interface FieldRules {
readonly decorated: ReadonlySet<string>;
/** Fields never written, whatever the author set. */
readonly omitted: ReadonlySet<string>;
+ /** Preserve an explicitly set value even when it equals the schema default.
*/
+ readonly keepDefaults?: boolean;
Review Comment:
Same here: drop the flag and the `field.default` comparison instead of
making it conditional.
```suggestion
```
##########
ts-sdk/src/coordinator/serde.ts:
##########
@@ -444,7 +449,7 @@ function applySchemaFields(
// `schedule` becomes `timetable`.
if (field.virtual || rules.omitted.has(field.key)) continue;
const value = values[name];
- if (value === undefined || value === field.default) continue;
+ if (value === undefined || (!rules.keepDefaults && value ===
field.default)) continue;
Review Comment:
```suggestion
if (value === undefined) continue;
```
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -2179,6 +2179,32 @@ def fill_config_defaults(cls, serialized_obj: dict[str,
Any]) -> None:
if key not in dag:
dag[key] = get(section, option)
+ # Native runtimes cannot read Airflow's task defaults. Supply only the
config-backed
+ # fields the Python serializer exposes through client_defaults, so
explicit values in
+ # the task itself (including values equal to schema defaults) retain
precedence.
+ task_default_keys = {
+ "owner",
+ "queue",
+ "retries",
+ "retry_delay",
+ "weight_rule",
+ "execution_timeout",
+ "email_on_failure",
+ "email_on_retry",
+ }
+ task_defaults = {
+ key: value
+ for key, value in
OperatorSerialization.generate_client_defaults().items()
+ if key in task_default_keys
+ }
Review Comment:
The whitelist is a second copy of knowledge `generate_client_defaults()`
already has: it drops every `OPERATOR_DEFAULTS` value equal to its schema
default, which today leaves exactly these eight keys (`pool_slots` and
`priority_weight` are hardcoded `1`, `allow_nested_operators` is `True` on both
sides, the rest are `None` or empty). Keeping the list means a native Dag
silently diverges from a Python Dag the day a new config-backed task default
lands in `OPERATOR_DEFAULTS`.
```suggestion
# Native runtimes cannot read Airflow's task defaults, so supply the
table a Python Dag
# carries. generate_client_defaults() already drops every value
equal to its schema
# default, which leaves exactly the config-backed fields.
task_defaults = OperatorSerialization.generate_client_defaults()
```
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -2179,6 +2179,32 @@ def fill_config_defaults(cls, serialized_obj: dict[str,
Any]) -> None:
if key not in dag:
dag[key] = get(section, option)
+ # Native runtimes cannot read Airflow's task defaults. Supply only the
config-backed
+ # fields the Python serializer exposes through client_defaults, so
explicit values in
+ # the task itself (including values equal to schema defaults) retain
precedence.
+ task_default_keys = {
+ "owner",
+ "queue",
+ "retries",
+ "retry_delay",
+ "weight_rule",
+ "execution_timeout",
+ "email_on_failure",
+ "email_on_retry",
+ }
+ task_defaults = {
+ key: value
+ for key, value in
OperatorSerialization.generate_client_defaults().items()
+ if key in task_default_keys
+ }
+ if task_defaults:
+ client_defaults = serialized_obj.setdefault("client_defaults", {})
+ if isinstance(client_defaults, dict):
+ existing_task_defaults = client_defaults.setdefault("tasks",
{})
+ if isinstance(existing_task_defaults, dict):
Review Comment:
No need the `if task_defaults:` and `if isinstance` here, we can assert the
client_defaults is correct as they're Airflow core module.
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -5466,6 +5466,77 @@ def test_keeps_a_setting_the_dag_sets(self):
assert data == before
+ def test_fills_unset_task_defaults_from_config(self, monkeypatch):
+ task_defaults = {
+ "owner": "configured-owner",
+ "queue": "celery_q",
+ "retries": 3,
+ "retry_delay": 240.0,
+ "weight_rule": "upstream",
+ "execution_timeout": 600.0,
+ "email_on_failure": False,
+ "email_on_retry": True,
+ "pool": "default_pool",
+ "priority_weight": 10,
+ }
+ monkeypatch.setattr(
+ OperatorSerialization,
+ "generate_client_defaults",
+ classmethod(lambda _: task_defaults),
+ )
+ task = {"__type": "operator", "__var": {"task_id": "explicit",
"language": "typescript", "retries": 0}}
+ data = {"__version": 3, "dag": {"dag_id": "native_dag", "tasks":
[task]}}
+
+ DagSerialization.fill_config_defaults(data)
+
+ assert data["client_defaults"] == {
+ "tasks": {
+ "owner": "configured-owner",
+ "queue": "celery_q",
+ "retries": 3,
+ "retry_delay": 240.0,
+ "weight_rule": "upstream",
+ "execution_timeout": 600.0,
+ "email_on_failure": False,
+ "email_on_retry": True,
+ }
+ }
+ assert data["dag"]["tasks"][0]["__var"]["retries"] == 0
+
+ def test_merges_task_defaults_without_replacing_explicit_defaults(self,
monkeypatch):
+ monkeypatch.setattr(
+ OperatorSerialization,
+ "generate_client_defaults",
+ classmethod(lambda _: {"retries": 3, "queue": "celery_q"}),
+ )
+ data = {
+ "__version": 3,
+ "dag": {"dag_id": "native_dag", "tasks": [{"__type": "operator",
"__var": {"task_id": "explicit", "language": "java"}}]},
+ "client_defaults": {"tasks": {"retries": 1}},
+ }
+
+ DagSerialization.fill_config_defaults(data)
+
+ assert data["client_defaults"] == {"tasks": {"retries": 1, "queue":
"celery_q"}}
+
+ def test_fills_task_defaults_for_go_sdk_payloads_too(self, monkeypatch):
+ monkeypatch.setattr(
+ OperatorSerialization,
+ "generate_client_defaults",
+ classmethod(lambda _: {"retries": 3, "queue": "celery_q"}),
+ )
+ data = {
+ "__version": 3,
+ "dag": {
+ "dag_id": "native_dag",
+ "tasks": [{"__type": "operator", "__var": {"task_id": "task",
"language": "go"}}],
+ },
+ }
+
+ DagSerialization.fill_config_defaults(data)
+
+ assert data["client_defaults"] == {"tasks": {"retries": 3, "queue":
"celery_q"}}
+
Review Comment:
We should also add a new test that define a real Dag with some operators
like in https://github.com/apache/airflow/pull/74869 to exercise the set
default behavior.
##########
scripts/ci/lang_sdk_serialization/compare.py:
##########
@@ -206,16 +206,17 @@ def compare_fields(
"""
Compare two serialized objects key by key.
- A key the SDK leaves out is fine when Python wrote its schema default.
Python keeps such a value
- when its ``client_defaults`` table disagrees with the schema, a table a
language SDK is never sent,
- and Airflow reads a missing field as its default anyway.
+ A key either side leaves out is fine when the value the other side writes
equals the schema
+ default. Python may omit a value matching its ``client_defaults`` table,
and native SDKs now preserve
+ explicitly set task values even when they equal the schema default.
Review Comment:
`compare_dag` still passes `{}` as the Dag-level defaults, so once the Dag
path writes explicit defaults too, a `fail_fast: false` the author set is
reported as a key Python does not write. It needs a Dag defaults table built
the way `get_task_defaults()` builds the operator one.
```suggestion
A key either side leaves out is fine when the value the other side
writes equals the schema
default: Python omits a value matching its ``client_defaults`` table,
and an SDK writes every
field the author set, including one that happens to equal the default.
```
##########
airflow-core/adr/lang-sdk/0004-dag-parsing.md:
##########
@@ -231,6 +231,17 @@ The language runtime must produce a `DagFileParsingResult`
that matches Python A
| `@continuous` | `airflow.timetables.simple.ContinuousTimetable` | `{}` |
| cron expr | `airflow.timetables.trigger.CronTriggerTimetable` |
`{expression, timezone, interval, run_immediately}` |
+**Task config defaults:** A runtime writes every task field the author
explicitly set, even when
+its value equals the schema default. A missing task field is filled from the
optional
+`client_defaults.tasks` table. Because language runtimes cannot read Airflow's
configuration,
+core supplies this table from its own config when the payload does not already
include it. Only
+config-backed task fields are supplied (`owner`, `queue`, `retries`,
`retry_delay`,
+`weight_rule`, `execution_timeout`, `email_on_failure`, and `email_on_retry`).
A task field
+present in the serialized task takes precedence, so an explicit `retries=0` is
not replaced by
+a non-zero `[core] default_task_retries`. The Go SDK also preserves explicit
schema-default values
+when its spec represents them as set; its non-pointer scalar zero values are
treated as unset and
+receive the configured default.
Review Comment:
Worth stating as one rule rather than a Go exception.
```suggestion
**Task config defaults:** A runtime writes every task field the author set
and leaves out the
ones they did not; a value equal to the schema default is still written. A
missing task field is
filled from the optional `client_defaults.tasks` table, which core supplies
from its own config
because a language runtime cannot read it. Only config-backed task fields
are supplied (`owner`,
`queue`, `retries`, `retry_delay`, `weight_rule`, `execution_timeout`,
`email_on_failure` and
`email_on_retry`). A field present in the serialized task wins over the
table, so an explicit
`retries=0` is not replaced by a non-zero `[core] default_task_retries`. In
Go a non-pointer
scalar at its zero value counts as unset and takes the configured default; a
pointer field
expresses an explicit zero.
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
