kaxil commented on code in PR #69864: URL: https://github.com/apache/airflow/pull/69864#discussion_r4032039057
########## airflow-core/src/airflow/serialization/dag_version_diff.py: ########## @@ -0,0 +1,1281 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +Observed-state diffs for serialized Dag payloads. + +Diff schema v1 -- the wire format every entry point here returns. +:func:`build_serialized_dag_diff` and :func:`build_unavailable_dag_diff` both return one +dictionary in the shape described below, and ``DIFF_SCHEMA_VERSION`` is the version of that +shape: bump it whenever a client-observable part of this contract changes. + +Top-level keys, all always present: + +* ``diff_schema_version`` -- ``int``, the value of ``DIFF_SCHEMA_VERSION``. +* ``serialized_dag_schema_versions`` -- ``{"base": int | None, "target": int | None}``, the + ``__version`` each stored payload carried. ``None`` means the key was absent or not an + integer; ``bool`` is rejected, so ``__version: true`` reads as ``None`` and never as + version 1. A version in ``SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS`` is upgraded to the + newest supported version before anything is compared. +* ``mode`` -- ``"observed_state"`` when the two payloads were compared, ``"unavailable"`` + when no comparison happened. +* ``changes`` -- the change records described below, ordered by the deterministic walk. + Always empty when ``mode`` is ``"unavailable"``. +* ``truncated`` -- ``True`` when more underlying changes exist than ``max_changes`` allowed. + Always ``False`` when ``mode`` is ``"unavailable"``. +* ``values`` -- ``{"status": "available" | "unavailable"}``. ``"available"`` only for an + ``"observed_state"`` result whose caller authorized value disclosure, so + ``result["values"]["status"]`` is safe to read on every result of every entry point. + +``unavailable_reason`` is the one conditional key: present exactly when ``mode`` is +``"unavailable"``. :func:`build_serialized_dag_diff` produces: + +* ``serialized_dag_missing`` -- one side has no stored payload. +* ``serialized_dag_schema_version_missing`` -- one side carries no usable ``__version``. +* ``unsupported_serialized_dag_schema_version:<n>`` -- ``<n>`` is the first unsupported + version found, base before target. +* ``serialized_dag_canonicalization_failed`` -- a payload could not be normalized (malformed + structure, an unsupported ``client_defaults`` section, an unencodable value). +* ``serialized_dag_recursion_limit_exceeded`` -- the comparison walk ran too deep. +* ``serialized_dag_json_encoding_failed`` -- a value could not be encoded as canonical JSON. + +:meth:`~airflow.models.dag_version.DagVersion.get_diff` reuses +``serialized_dag_decode_failed``, for a stored payload that will not decompress or parse, and +``deadline_alert_missing``, for a payload that references a deadline alert row that is gone. +:func:`build_unavailable_dag_diff` passes its caller's reason through unchanged, so a new +caller extends this list rather than inventing an undocumented value. + +Each change record carries these keys in both disclosure modes: + +* ``path`` -- a JSON-Pointer-style path into a synthetic document with two roots: ``/dag`` + (the canonicalized serialized Dag, with ``__version`` dropped and client defaults folded + into the tasks) and ``/provenance`` (the provenance mapping the caller supplied -- + ``bundle_name``, ``bundle_version`` and ``version_data`` from ``get_diff``). ``~`` and + ``/`` inside a component are escaped as ``~0`` and ``~1``. +* ``operation`` -- ``"added"``, ``"removed"`` or ``"changed"``. +* ``category`` -- a :data:`DiffCategory`: ``task`` for task definitions and the Dag fields + that shape task execution, ``dependency`` for ``dag_dependencies`` and downstream task + ids, ``schedule`` for timetable, dates and concurrency, ``param`` for params, + ``asset`` for task inlets and outlets, ``deadline`` for deadline alerts, + ``callback`` for callback presence, ``metadata`` for descriptive and display fields, + ``authorization`` for ``access_control``, ``provenance`` for everything under + ``/provenance`` plus the file and bundle locators, and ``unknown`` for anything + unclassified. ``_DIFF_V1_DAG_FIELD_CATEGORIES`` is the authoritative Dag-field mapping; + an unclassified task field falls back to ``task``, which over-reports impact. + ``default_args`` is the one Dag field classified per key rather than as a whole, because its + keys are the operator fields they will be applied to: an allowlisted key takes that field's + category (``owner`` is ``metadata``, ``retries`` is ``task``), and the aggregate record for + the rest keeps the conservative ``task``. Only a stored value that is not a readable dict + envelope classifies as ``task`` as a whole. +* ``impact`` -- a :data:`DiffImpact` derived from ``category``: ``provenance``, ``metadata`` + and ``authorization`` carry through unchanged, every operational category becomes + ``execution``, and ``unknown`` stays ``unknown``. +* ``occurrence_count`` -- how many underlying changes the record stands for. + +A caller that authorized values (``include_values=True``, ``values.status`` ``"available"``) +additionally gets: + +* ``before_digest`` / ``after_digest`` -- ``"sha256:<hex>"`` over the canonical JSON of that + side, or ``None`` when that side is missing. Both keys are always present. +* ``before_value`` / ``after_value`` -- the canonicalized value, which is the stored one after + schema and client defaults are folded in and params are normalized. ``before_value`` is absent from + an ``added`` record and ``after_value`` from a ``removed`` one, which is how a missing side + is told apart from a stored ``null``. + +``path`` and ``occurrence_count`` also differ between the modes. A redacted record reports the +public path: members of keyed collections (``/dag/tasks``, ``/dag/dag_dependencies``, +``/dag/tags``, ``/dag/allowed_run_types`` and task group children) collapse to ``*`` instead +of naming a task, tag or group, and records sharing a public path and operation are merged +into one whose ``occurrence_count`` counts the merged changes. An authorized record keeps the +identifying component and always has an ``occurrence_count`` of 1. Fields outside the v1 task, +task group and ``default_args`` allowlists are aggregated into a single ``custom_fields`` record +in both modes, so neither a private serializer field nor a user-chosen ``default_args`` key is +ever named by either -- ``/dag/default_args`` names only allowlisted operator field names, which +is why its keys can be classified at all. A top-level key outside the v1 root allowlist is +likewise reported as ``/custom_fields`` when redacted, while an authorized record keeps its real +root path. + +``max_changes`` bounds the result rather than the traversal: both modes walk the same shape and +reach the same changes, so authorization never decides which changes exist, and the public paths +either mode reports -- and the order of their first occurrence -- are the same. What the bound +admits does differ. An authorized record stands for one change, so the bound caps those directly. +A redacted record stands for every change sharing its public path and operation, and a change +reaching a record already in the result costs no new one, so redacted counting continues past the +bound until a change needs a record the bound will not allow. The same payload at the same bound +can therefore report more occurrences, and a different ``truncated``, when values are withheld. + +``truncated`` reports that such a change was dropped. When it is ``False`` every changed path +is present with an exact ``occurrence_count``. When it is ``True`` the walk stopped early: some +paths are absent rather than unchanged, and every ``occurrence_count`` is a lower bound, since +occurrences after the stop were never reached. + +Typed dictionaries for the result and the change record are deliberately deferred -- the +record shape varies with disclosure mode, so modelling it belongs with the REST layer that +exposes it. :data:`DiffCategory` and :data:`DiffImpact` are stable enough to generate enums +from today. +""" + +from __future__ import annotations + +import copy +import hashlib +import json +from collections.abc import Callable, Mapping, MutableMapping +from contextlib import suppress +from dataclasses import dataclass +from datetime import timedelta +from enum import Enum +from typing import Any, Literal, NamedTuple + +import structlog + +from airflow.serialization.definitions.baseoperator import SerializedBaseOperator +from airflow.serialization.definitions.mappedoperator import SerializedMappedOperator +from airflow.serialization.serialized_objects import ( + _DAG_CALLBACK_FIELDS, + _OPERATOR_TIMEDELTA_FIELDS, + DagSerialization, + OperatorSerialization, +) + +log = structlog.get_logger(__name__) + +DIFF_SCHEMA_VERSION = 1 +DEFAULT_MAX_CHANGES = 500 +MAX_ALLOWED_CHANGES = 5000 +SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS = frozenset((1, 2, 3)) + +DiffCategory = Literal[ + "asset", + "authorization", + "callback", + "deadline", + "dependency", + "metadata", + "param", + "provenance", + "schedule", + "task", + "unknown", +] +DiffImpact = Literal["authorization", "execution", "metadata", "provenance", "unknown"] + +_ORDER_INSENSITIVE_LIST_PATHS = { + ("dag", "tags"), + ("dag", "allowed_run_types"), +} +_KEYED_COLLECTION_PATHS = { + ("dag", "tasks"), + ("dag", "dag_dependencies"), + *_ORDER_INSENSITIVE_LIST_PATHS, +} +# Canonicalization folds away __version and client_defaults and adds provenance, so these are the +# only top-level sections the walk is allowed to name. +_DIFF_V1_PUBLIC_ROOT_FIELDS = frozenset({"dag", "provenance"}) +_CUSTOM_TASK_FIELDS_PATH_COMPONENT = "custom_fields" +# This allowlist is part of diff schema v1. Serializer schema changes must not +# silently change the paths visible to callers of the diff API. +_DIFF_V1_PUBLIC_TASK_FIELDS = frozenset( + { + "__type", + "_can_skip_downstream", + "_disallow_kwargs_override", + "_expand_input_attr", + "_is_empty", + "_is_mapped", + "_is_sensor", + "_logger_name", + "_needs_expansion", + "_operator_extra_links", + "_operator_name", + "_task_display_name", + "_task_module", + "allow_nested_operators", + "depends_on_past", + "do_xcom_push", + "doc", + "doc_json", + "doc_md", + "doc_rst", + "doc_yaml", + "downstream_task_ids", + "email", + "email_on_failure", + "email_on_retry", + "end_date", + "execution_timeout", + "executor", + "executor_config", + "expand_input", + "has_on_execute_callback", + "has_on_failure_callback", + "has_on_retry_callback", + "has_on_skipped_callback", + "has_on_success_callback", + "has_retry_policy", + "ignore_first_depends_on_past", + "inlets", + "is_setup", + "is_stub", + "is_teardown", + "map_index_template", + "max_active_tis_per_dag", + "max_active_tis_per_dagrun", + "max_retry_delay", + "multiple_outputs", + "on_failure_fail_dagrun", + "op_kwargs_expand_input", + "outlets", + "owner", + "params", + "partial_kwargs", + "pool", + "pool_slots", + "priority_weight", + "python_callable_name", + "queue", + "render_template_as_native_obj", + "reschedule", + "resources", + "retries", + "retry_delay", + "retry_exponential_backoff", + "run_as_user", + "start_date", + "start_from_trigger", + "start_trigger_args", + "task_id", + "task_type", + "template_ext", + "template_fields", + "template_fields_renderers", + "trigger_rule", + "ui_color", + "ui_fgcolor", + "wait_for_downstream", + "wait_for_past_depends_before_skipping", + "weight_rule", + } +) +_DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS = frozenset({"_arg_bindings"}) +# Task-field categories used by _get_category. +_DIFF_V1_TASK_ASSET_FIELDS = frozenset({"inlets", "outlets"}) +_DIFF_V1_TASK_PARAM_FIELDS = frozenset({"params"}) +_DIFF_V1_TASK_DEPENDENCY_FIELDS = frozenset({"downstream_task_ids"}) +_DIFF_V1_TASK_METADATA_FIELDS = frozenset( + { + "_operator_name", + "doc", + "doc_json", + "doc_md", + "doc_rst", + "doc_yaml", + "owner", + "ui_color", + "ui_fgcolor", + "_task_display_name", + "task_display_name", + } +) +_DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS = _DIFF_V1_PUBLIC_TASK_FIELDS | {"task_display_name"} +_DEFAULT_ARGS_PATH = ("dag", "default_args") +_RETRY_BACKOFF_FIELD = "retry_exponential_backoff" +# Dag fields the schema gives no default for, added after the first serializer versions. +_ABSENT_AS_NULL_DAG_FIELDS = ("allowed_run_types", "deadline") +# These hydrate to sets (_deserialize_operator_field for a task, TaskGroup.*_ids.update for a group), +# so a reordered stored list means the same edges and must not read as a dependency change. +_SET_VALUED_ID_FIELDS = frozenset( + {"downstream_task_ids", "upstream_task_ids", "downstream_group_ids", "upstream_group_ids"} +) +# A Dag's default_args holds the same operator __init__ kwargs partial_kwargs holds -- the +# serializer rewrites _HAS_FLAG_FIELDS in both -- so it reuses that allowlist. "__type" names the +# stored value's own dict envelope rather than a key inside it. +_DIFF_V1_PUBLIC_DEFAULT_ARGS_FIELDS = _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS - {"__type"} +# Classify every Dag schema field explicitly so new fields require a policy decision. +_DIFF_V1_DAG_FIELD_CATEGORIES: dict[str, DiffCategory] = { + "_concurrency": "schedule", + "_processor_dags_folder": "provenance", + "access_control": "authorization", + "allowed_run_types": "schedule", + "bundle_name": "provenance", + "catchup": "schedule", + "dag_dependencies": "dependency", + "dag_display_name": "metadata", + "dag_id": "metadata", + "dagrun_timeout": "schedule", + "deadline": "deadline", + # Only reached when the stored value is not the dict envelope _collect_default_args_changes + # walks per key; an unreadable envelope keeps the conservative impact of its widest key. + "default_args": "task", + "description": "metadata", + # Decides whether a run pins a bundle version, so it changes which code later runs execute + # and whether triggering a historical version is allowed at all. + "disable_bundle_versioning": "task", + "doc_md": "metadata", + "edge_info": "metadata", + "end_date": "schedule", + "fail_fast": "schedule", + "fileloc": "provenance", + "has_on_failure_callback": "callback", + "has_on_success_callback": "callback", + "is_paused_upon_creation": "schedule", + "max_active_runs": "schedule", + "max_active_tasks": "schedule", + "max_consecutive_failed_dag_runs": "schedule", + "owner_links": "metadata", + "params": "param", + "relative_fileloc": "provenance", + "render_template_as_native_obj": "task", + "rerun_with_latest_version": "task", + "start_date": "schedule", + "tags": "metadata", + "task_group": "task", + "tasks": "task", + "timetable": "schedule", + "timezone": "schedule", +} +# Fields dropped from the current schema that a stored payload can still carry. Every entry must +# survive _canonicalize_payload_v1, or it classifies nothing. +_DIFF_V1_LEGACY_DAG_FIELD_CATEGORIES: dict[str, DiffCategory] = { + "fail_stop": "schedule", + "on_failure_callback": "callback", + "on_success_callback": "callback", + "schedule": "schedule", +} +_RECURSIVE_MAPPING_PATHS = { + (), + ("dag",), + ("provenance",), + *_KEYED_COLLECTION_PATHS, +} +_DIFF_V1_TASK_GROUP_METADATA_FIELDS = frozenset( + {"group_display_name", "tooltip", "doc_md", "ui_color", "ui_fgcolor"} +) +_DIFF_V1_PUBLIC_TASK_GROUP_FIELDS = _DIFF_V1_TASK_GROUP_METADATA_FIELDS | { + "_group_id", + "prefix_group_id", + "children", + "upstream_group_ids", + "downstream_group_ids", + "upstream_task_ids", + "downstream_task_ids", + "expand_input", + "is_mapped", +} + + +def build_unavailable_dag_diff( + *, + base_data: dict[str, Any] | None, + target_data: dict[str, Any] | None, + reason: str, +) -> dict[str, Any]: + """Report a known unavailable reason without comparing the stored payloads.""" + return _mark_unavailable( + _build_diff_result(_get_schema_version(base_data), _get_schema_version(target_data)), reason + ) + + +def build_serialized_dag_diff( + *, + base_data: dict[str, Any] | None, + target_data: dict[str, Any] | None, + base_provenance: Mapping[str, Any] | None = None, + target_provenance: Mapping[str, Any] | None = None, + include_values: bool = False, + max_changes: int = DEFAULT_MAX_CHANGES, +) -> dict[str, Any]: + """ + Compare two serialized Dag payloads and their provenance deterministically. + + Callers must authorize disclosure of the entire serialized payload, including + access-control roles and permissions, before setting ``include_values=True``. + This exposes canonicalized values, digests, and identifying path components. + + ``max_changes`` limits underlying changes admitted to the result. Redacted changes + with the same public path and operation share a record. Repeats can be counted past + the limit until a change needing a new record stops the walk. When ``truncated`` is + true, some paths are absent and occurrence counts are lower bounds. + + An ``unavailable`` result includes the reason comparison failed. + """ + validate_max_changes(max_changes) + + base_schema_version = _get_schema_version(base_data) + target_schema_version = _get_schema_version(target_data) + result = _build_diff_result(base_schema_version, target_schema_version) + + if base_data is None or target_data is None: + return _mark_unavailable(result, "serialized_dag_missing") + + if base_schema_version is None or target_schema_version is None: + return _mark_unavailable(result, "serialized_dag_schema_version_missing") + + unsupported_versions = [ + version + for version in (base_schema_version, target_schema_version) + if version not in SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS + ] + if unsupported_versions: + return _mark_unavailable( + result, f"unsupported_serialized_dag_schema_version:{unsupported_versions[0]}" + ) + + try: + base_document = _canonicalize_payload_v1(base_data) + target_document = _canonicalize_payload_v1(target_data) + base_document["provenance"] = _canonicalize_value(dict(base_provenance or {}), path=("provenance",)) + target_document["provenance"] = _canonicalize_value( + dict(target_provenance or {}), path=("provenance",) + ) + except (AttributeError, KeyError, OverflowError, RecursionError, TypeError, ValueError) as error: + log.warning( + "Serialized Dag diff canonicalization failed", + error_type=type(error).__name__, Review Comment: This logs the exception class and nothing else, so the messages added to make these failures loud do not survive. `_apply_task_defaults` raises `ValueError(f"unsupported client_defaults sections: {sorted(unknown_sections)}")` for exactly the case from two rounds ago, and `dag.tasks is not a list`, `task entry has no task_id` and `duplicate key ... in /dag/tasks` all land in this same handler. What an operator sees in every one of those cases is `Serialized Dag diff canonicalization failed base_schema_version=3 error_type=ValueError target_schema_version=3`. The tuple also catches `AttributeError`, `KeyError` and `TypeError`, so an engine bug arrives here too and reads identically to a new serializer section. `exc_info=error`, or a `reason=str(error)` field, tells them apart. ########## airflow-core/src/airflow/serialization/dag_version_diff.py: ########## @@ -0,0 +1,1281 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +Observed-state diffs for serialized Dag payloads. + +Diff schema v1 -- the wire format every entry point here returns. +:func:`build_serialized_dag_diff` and :func:`build_unavailable_dag_diff` both return one +dictionary in the shape described below, and ``DIFF_SCHEMA_VERSION`` is the version of that +shape: bump it whenever a client-observable part of this contract changes. + +Top-level keys, all always present: + +* ``diff_schema_version`` -- ``int``, the value of ``DIFF_SCHEMA_VERSION``. +* ``serialized_dag_schema_versions`` -- ``{"base": int | None, "target": int | None}``, the + ``__version`` each stored payload carried. ``None`` means the key was absent or not an + integer; ``bool`` is rejected, so ``__version: true`` reads as ``None`` and never as + version 1. A version in ``SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS`` is upgraded to the + newest supported version before anything is compared. +* ``mode`` -- ``"observed_state"`` when the two payloads were compared, ``"unavailable"`` + when no comparison happened. +* ``changes`` -- the change records described below, ordered by the deterministic walk. + Always empty when ``mode`` is ``"unavailable"``. +* ``truncated`` -- ``True`` when more underlying changes exist than ``max_changes`` allowed. + Always ``False`` when ``mode`` is ``"unavailable"``. +* ``values`` -- ``{"status": "available" | "unavailable"}``. ``"available"`` only for an + ``"observed_state"`` result whose caller authorized value disclosure, so + ``result["values"]["status"]`` is safe to read on every result of every entry point. + +``unavailable_reason`` is the one conditional key: present exactly when ``mode`` is +``"unavailable"``. :func:`build_serialized_dag_diff` produces: + +* ``serialized_dag_missing`` -- one side has no stored payload. +* ``serialized_dag_schema_version_missing`` -- one side carries no usable ``__version``. +* ``unsupported_serialized_dag_schema_version:<n>`` -- ``<n>`` is the first unsupported + version found, base before target. +* ``serialized_dag_canonicalization_failed`` -- a payload could not be normalized (malformed + structure, an unsupported ``client_defaults`` section, an unencodable value). +* ``serialized_dag_recursion_limit_exceeded`` -- the comparison walk ran too deep. +* ``serialized_dag_json_encoding_failed`` -- a value could not be encoded as canonical JSON. + +:meth:`~airflow.models.dag_version.DagVersion.get_diff` reuses +``serialized_dag_decode_failed``, for a stored payload that will not decompress or parse, and +``deadline_alert_missing``, for a payload that references a deadline alert row that is gone. +:func:`build_unavailable_dag_diff` passes its caller's reason through unchanged, so a new +caller extends this list rather than inventing an undocumented value. + +Each change record carries these keys in both disclosure modes: + +* ``path`` -- a JSON-Pointer-style path into a synthetic document with two roots: ``/dag`` + (the canonicalized serialized Dag, with ``__version`` dropped and client defaults folded + into the tasks) and ``/provenance`` (the provenance mapping the caller supplied -- + ``bundle_name``, ``bundle_version`` and ``version_data`` from ``get_diff``). ``~`` and + ``/`` inside a component are escaped as ``~0`` and ``~1``. +* ``operation`` -- ``"added"``, ``"removed"`` or ``"changed"``. +* ``category`` -- a :data:`DiffCategory`: ``task`` for task definitions and the Dag fields + that shape task execution, ``dependency`` for ``dag_dependencies`` and downstream task + ids, ``schedule`` for timetable, dates and concurrency, ``param`` for params, + ``asset`` for task inlets and outlets, ``deadline`` for deadline alerts, + ``callback`` for callback presence, ``metadata`` for descriptive and display fields, + ``authorization`` for ``access_control``, ``provenance`` for everything under + ``/provenance`` plus the file and bundle locators, and ``unknown`` for anything + unclassified. ``_DIFF_V1_DAG_FIELD_CATEGORIES`` is the authoritative Dag-field mapping; + an unclassified task field falls back to ``task``, which over-reports impact. + ``default_args`` is the one Dag field classified per key rather than as a whole, because its + keys are the operator fields they will be applied to: an allowlisted key takes that field's + category (``owner`` is ``metadata``, ``retries`` is ``task``), and the aggregate record for + the rest keeps the conservative ``task``. Only a stored value that is not a readable dict + envelope classifies as ``task`` as a whole. +* ``impact`` -- a :data:`DiffImpact` derived from ``category``: ``provenance``, ``metadata`` + and ``authorization`` carry through unchanged, every operational category becomes + ``execution``, and ``unknown`` stays ``unknown``. +* ``occurrence_count`` -- how many underlying changes the record stands for. + +A caller that authorized values (``include_values=True``, ``values.status`` ``"available"``) +additionally gets: + +* ``before_digest`` / ``after_digest`` -- ``"sha256:<hex>"`` over the canonical JSON of that + side, or ``None`` when that side is missing. Both keys are always present. +* ``before_value`` / ``after_value`` -- the canonicalized value, which is the stored one after + schema and client defaults are folded in and params are normalized. ``before_value`` is absent from + an ``added`` record and ``after_value`` from a ``removed`` one, which is how a missing side + is told apart from a stored ``null``. + +``path`` and ``occurrence_count`` also differ between the modes. A redacted record reports the +public path: members of keyed collections (``/dag/tasks``, ``/dag/dag_dependencies``, +``/dag/tags``, ``/dag/allowed_run_types`` and task group children) collapse to ``*`` instead +of naming a task, tag or group, and records sharing a public path and operation are merged +into one whose ``occurrence_count`` counts the merged changes. An authorized record keeps the +identifying component and always has an ``occurrence_count`` of 1. Fields outside the v1 task, +task group and ``default_args`` allowlists are aggregated into a single ``custom_fields`` record +in both modes, so neither a private serializer field nor a user-chosen ``default_args`` key is +ever named by either -- ``/dag/default_args`` names only allowlisted operator field names, which +is why its keys can be classified at all. A top-level key outside the v1 root allowlist is +likewise reported as ``/custom_fields`` when redacted, while an authorized record keeps its real +root path. + +``max_changes`` bounds the result rather than the traversal: both modes walk the same shape and +reach the same changes, so authorization never decides which changes exist, and the public paths +either mode reports -- and the order of their first occurrence -- are the same. What the bound +admits does differ. An authorized record stands for one change, so the bound caps those directly. +A redacted record stands for every change sharing its public path and operation, and a change +reaching a record already in the result costs no new one, so redacted counting continues past the +bound until a change needs a record the bound will not allow. The same payload at the same bound +can therefore report more occurrences, and a different ``truncated``, when values are withheld. + +``truncated`` reports that such a change was dropped. When it is ``False`` every changed path +is present with an exact ``occurrence_count``. When it is ``True`` the walk stopped early: some +paths are absent rather than unchanged, and every ``occurrence_count`` is a lower bound, since +occurrences after the stop were never reached. + +Typed dictionaries for the result and the change record are deliberately deferred -- the +record shape varies with disclosure mode, so modelling it belongs with the REST layer that +exposes it. :data:`DiffCategory` and :data:`DiffImpact` are stable enough to generate enums +from today. +""" + +from __future__ import annotations + +import copy +import hashlib +import json +from collections.abc import Callable, Mapping, MutableMapping +from contextlib import suppress +from dataclasses import dataclass +from datetime import timedelta +from enum import Enum +from typing import Any, Literal, NamedTuple + +import structlog + +from airflow.serialization.definitions.baseoperator import SerializedBaseOperator +from airflow.serialization.definitions.mappedoperator import SerializedMappedOperator +from airflow.serialization.serialized_objects import ( + _DAG_CALLBACK_FIELDS, + _OPERATOR_TIMEDELTA_FIELDS, + DagSerialization, + OperatorSerialization, +) + +log = structlog.get_logger(__name__) + +DIFF_SCHEMA_VERSION = 1 +DEFAULT_MAX_CHANGES = 500 +MAX_ALLOWED_CHANGES = 5000 +SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS = frozenset((1, 2, 3)) + +DiffCategory = Literal[ + "asset", + "authorization", + "callback", + "deadline", + "dependency", + "metadata", + "param", + "provenance", + "schedule", + "task", + "unknown", +] +DiffImpact = Literal["authorization", "execution", "metadata", "provenance", "unknown"] + +_ORDER_INSENSITIVE_LIST_PATHS = { + ("dag", "tags"), + ("dag", "allowed_run_types"), +} +_KEYED_COLLECTION_PATHS = { + ("dag", "tasks"), + ("dag", "dag_dependencies"), + *_ORDER_INSENSITIVE_LIST_PATHS, +} +# Canonicalization folds away __version and client_defaults and adds provenance, so these are the +# only top-level sections the walk is allowed to name. +_DIFF_V1_PUBLIC_ROOT_FIELDS = frozenset({"dag", "provenance"}) +_CUSTOM_TASK_FIELDS_PATH_COMPONENT = "custom_fields" +# This allowlist is part of diff schema v1. Serializer schema changes must not +# silently change the paths visible to callers of the diff API. +_DIFF_V1_PUBLIC_TASK_FIELDS = frozenset( + { + "__type", + "_can_skip_downstream", + "_disallow_kwargs_override", + "_expand_input_attr", + "_is_empty", + "_is_mapped", + "_is_sensor", + "_logger_name", + "_needs_expansion", + "_operator_extra_links", + "_operator_name", + "_task_display_name", + "_task_module", + "allow_nested_operators", + "depends_on_past", + "do_xcom_push", + "doc", + "doc_json", + "doc_md", + "doc_rst", + "doc_yaml", + "downstream_task_ids", + "email", + "email_on_failure", + "email_on_retry", + "end_date", + "execution_timeout", + "executor", + "executor_config", + "expand_input", + "has_on_execute_callback", + "has_on_failure_callback", + "has_on_retry_callback", + "has_on_skipped_callback", + "has_on_success_callback", + "has_retry_policy", + "ignore_first_depends_on_past", + "inlets", + "is_setup", + "is_stub", + "is_teardown", + "map_index_template", + "max_active_tis_per_dag", + "max_active_tis_per_dagrun", + "max_retry_delay", + "multiple_outputs", + "on_failure_fail_dagrun", + "op_kwargs_expand_input", + "outlets", + "owner", + "params", + "partial_kwargs", + "pool", + "pool_slots", + "priority_weight", + "python_callable_name", + "queue", + "render_template_as_native_obj", + "reschedule", + "resources", + "retries", + "retry_delay", + "retry_exponential_backoff", + "run_as_user", + "start_date", + "start_from_trigger", + "start_trigger_args", + "task_id", + "task_type", + "template_ext", + "template_fields", + "template_fields_renderers", + "trigger_rule", + "ui_color", + "ui_fgcolor", + "wait_for_downstream", + "wait_for_past_depends_before_skipping", + "weight_rule", + } +) +_DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS = frozenset({"_arg_bindings"}) +# Task-field categories used by _get_category. +_DIFF_V1_TASK_ASSET_FIELDS = frozenset({"inlets", "outlets"}) +_DIFF_V1_TASK_PARAM_FIELDS = frozenset({"params"}) +_DIFF_V1_TASK_DEPENDENCY_FIELDS = frozenset({"downstream_task_ids"}) +_DIFF_V1_TASK_METADATA_FIELDS = frozenset( + { + "_operator_name", + "doc", + "doc_json", + "doc_md", + "doc_rst", + "doc_yaml", + "owner", + "ui_color", + "ui_fgcolor", + "_task_display_name", + "task_display_name", + } +) +_DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS = _DIFF_V1_PUBLIC_TASK_FIELDS | {"task_display_name"} +_DEFAULT_ARGS_PATH = ("dag", "default_args") +_RETRY_BACKOFF_FIELD = "retry_exponential_backoff" +# Dag fields the schema gives no default for, added after the first serializer versions. +_ABSENT_AS_NULL_DAG_FIELDS = ("allowed_run_types", "deadline") +# These hydrate to sets (_deserialize_operator_field for a task, TaskGroup.*_ids.update for a group), +# so a reordered stored list means the same edges and must not read as a dependency change. +_SET_VALUED_ID_FIELDS = frozenset( + {"downstream_task_ids", "upstream_task_ids", "downstream_group_ids", "upstream_group_ids"} +) +# A Dag's default_args holds the same operator __init__ kwargs partial_kwargs holds -- the +# serializer rewrites _HAS_FLAG_FIELDS in both -- so it reuses that allowlist. "__type" names the +# stored value's own dict envelope rather than a key inside it. +_DIFF_V1_PUBLIC_DEFAULT_ARGS_FIELDS = _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS - {"__type"} +# Classify every Dag schema field explicitly so new fields require a policy decision. +_DIFF_V1_DAG_FIELD_CATEGORIES: dict[str, DiffCategory] = { + "_concurrency": "schedule", + "_processor_dags_folder": "provenance", + "access_control": "authorization", + "allowed_run_types": "schedule", + "bundle_name": "provenance", + "catchup": "schedule", + "dag_dependencies": "dependency", + "dag_display_name": "metadata", + "dag_id": "metadata", + "dagrun_timeout": "schedule", + "deadline": "deadline", + # Only reached when the stored value is not the dict envelope _collect_default_args_changes + # walks per key; an unreadable envelope keeps the conservative impact of its widest key. + "default_args": "task", + "description": "metadata", + # Decides whether a run pins a bundle version, so it changes which code later runs execute + # and whether triggering a historical version is allowed at all. + "disable_bundle_versioning": "task", + "doc_md": "metadata", + "edge_info": "metadata", + "end_date": "schedule", + "fail_fast": "schedule", + "fileloc": "provenance", + "has_on_failure_callback": "callback", + "has_on_success_callback": "callback", + "is_paused_upon_creation": "schedule", + "max_active_runs": "schedule", + "max_active_tasks": "schedule", + "max_consecutive_failed_dag_runs": "schedule", + "owner_links": "metadata", + "params": "param", + "relative_fileloc": "provenance", + "render_template_as_native_obj": "task", + "rerun_with_latest_version": "task", + "start_date": "schedule", + "tags": "metadata", + "task_group": "task", + "tasks": "task", + "timetable": "schedule", + "timezone": "schedule", +} +# Fields dropped from the current schema that a stored payload can still carry. Every entry must +# survive _canonicalize_payload_v1, or it classifies nothing. +_DIFF_V1_LEGACY_DAG_FIELD_CATEGORIES: dict[str, DiffCategory] = { + "fail_stop": "schedule", + "on_failure_callback": "callback", + "on_success_callback": "callback", + "schedule": "schedule", +} +_RECURSIVE_MAPPING_PATHS = { + (), + ("dag",), + ("provenance",), + *_KEYED_COLLECTION_PATHS, +} +_DIFF_V1_TASK_GROUP_METADATA_FIELDS = frozenset( + {"group_display_name", "tooltip", "doc_md", "ui_color", "ui_fgcolor"} +) +_DIFF_V1_PUBLIC_TASK_GROUP_FIELDS = _DIFF_V1_TASK_GROUP_METADATA_FIELDS | { + "_group_id", + "prefix_group_id", + "children", + "upstream_group_ids", + "downstream_group_ids", + "upstream_task_ids", + "downstream_task_ids", + "expand_input", + "is_mapped", +} + + +def build_unavailable_dag_diff( + *, + base_data: dict[str, Any] | None, + target_data: dict[str, Any] | None, + reason: str, +) -> dict[str, Any]: + """Report a known unavailable reason without comparing the stored payloads.""" + return _mark_unavailable( + _build_diff_result(_get_schema_version(base_data), _get_schema_version(target_data)), reason + ) + + +def build_serialized_dag_diff( + *, + base_data: dict[str, Any] | None, + target_data: dict[str, Any] | None, + base_provenance: Mapping[str, Any] | None = None, + target_provenance: Mapping[str, Any] | None = None, + include_values: bool = False, + max_changes: int = DEFAULT_MAX_CHANGES, +) -> dict[str, Any]: + """ + Compare two serialized Dag payloads and their provenance deterministically. + + Callers must authorize disclosure of the entire serialized payload, including + access-control roles and permissions, before setting ``include_values=True``. + This exposes canonicalized values, digests, and identifying path components. + + ``max_changes`` limits underlying changes admitted to the result. Redacted changes + with the same public path and operation share a record. Repeats can be counted past + the limit until a change needing a new record stops the walk. When ``truncated`` is + true, some paths are absent and occurrence counts are lower bounds. + + An ``unavailable`` result includes the reason comparison failed. + """ + validate_max_changes(max_changes) + + base_schema_version = _get_schema_version(base_data) + target_schema_version = _get_schema_version(target_data) + result = _build_diff_result(base_schema_version, target_schema_version) + + if base_data is None or target_data is None: + return _mark_unavailable(result, "serialized_dag_missing") + + if base_schema_version is None or target_schema_version is None: + return _mark_unavailable(result, "serialized_dag_schema_version_missing") + + unsupported_versions = [ + version + for version in (base_schema_version, target_schema_version) + if version not in SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS + ] + if unsupported_versions: + return _mark_unavailable( + result, f"unsupported_serialized_dag_schema_version:{unsupported_versions[0]}" + ) + + try: + base_document = _canonicalize_payload_v1(base_data) + target_document = _canonicalize_payload_v1(target_data) + base_document["provenance"] = _canonicalize_value(dict(base_provenance or {}), path=("provenance",)) + target_document["provenance"] = _canonicalize_value( + dict(target_provenance or {}), path=("provenance",) + ) + except (AttributeError, KeyError, OverflowError, RecursionError, TypeError, ValueError) as error: + log.warning( + "Serialized Dag diff canonicalization failed", + error_type=type(error).__name__, + base_schema_version=base_schema_version, + target_schema_version=target_schema_version, + ) + return _mark_unavailable(result, "serialized_dag_canonicalization_failed") + + collector = _ChangeCollector(max_changes=max_changes, include_values=include_values) + try: + _collect_changes(base_document, target_document, path=(), collector=collector) + except RecursionError: + log.warning( + "Serialized Dag diff recursion limit exceeded", + base_schema_version=base_schema_version, + target_schema_version=target_schema_version, + ) + return _mark_unavailable(result, "serialized_dag_recursion_limit_exceeded") Review Comment: I do not think this reason can reach a caller, and the mechanism I gave for it last round was wrong. Canonicalization runs first over a strict superset of what the walk recurses into, so it exhausts the stack first in every shape I could build. Swept at this HEAD in breeze at the default limit of 1000: a deeply nested `executor_config` returns `observed_state` through depth 488 and `serialized_dag_canonicalization_failed` from 490 on, and nested task groups, the one place the walk itself recurses without bound, are clean through depth 164 and `serialized_dag_canonicalization_failed` from 165 on. There is no band where the walk fails first. Both recursion tests mock the function they exercise (`_collect_changes` on line 1896, `_canonicalize_payload_v1` on line 1873), so neither observed which reason a real deep payload produces. Deep payloads still degrade correctly, so this is about the wire contract rather than behaviour: line 53 lists `serialized_dag_recursion_limit_exceeded` among the reasons a client can receive, and the CLI and endpoint PRs will enumerate it from there. Either drop the handler with the bullet, or keep it and say in the docstring that it is defence in depth rather than a reachable reason. ########## airflow-core/src/airflow/serialization/dag_version_diff.py: ########## @@ -0,0 +1,1281 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +Observed-state diffs for serialized Dag payloads. + +Diff schema v1 -- the wire format every entry point here returns. +:func:`build_serialized_dag_diff` and :func:`build_unavailable_dag_diff` both return one +dictionary in the shape described below, and ``DIFF_SCHEMA_VERSION`` is the version of that +shape: bump it whenever a client-observable part of this contract changes. + +Top-level keys, all always present: + +* ``diff_schema_version`` -- ``int``, the value of ``DIFF_SCHEMA_VERSION``. +* ``serialized_dag_schema_versions`` -- ``{"base": int | None, "target": int | None}``, the + ``__version`` each stored payload carried. ``None`` means the key was absent or not an + integer; ``bool`` is rejected, so ``__version: true`` reads as ``None`` and never as + version 1. A version in ``SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS`` is upgraded to the + newest supported version before anything is compared. +* ``mode`` -- ``"observed_state"`` when the two payloads were compared, ``"unavailable"`` + when no comparison happened. +* ``changes`` -- the change records described below, ordered by the deterministic walk. + Always empty when ``mode`` is ``"unavailable"``. +* ``truncated`` -- ``True`` when more underlying changes exist than ``max_changes`` allowed. + Always ``False`` when ``mode`` is ``"unavailable"``. +* ``values`` -- ``{"status": "available" | "unavailable"}``. ``"available"`` only for an + ``"observed_state"`` result whose caller authorized value disclosure, so + ``result["values"]["status"]`` is safe to read on every result of every entry point. + +``unavailable_reason`` is the one conditional key: present exactly when ``mode`` is +``"unavailable"``. :func:`build_serialized_dag_diff` produces: + +* ``serialized_dag_missing`` -- one side has no stored payload. +* ``serialized_dag_schema_version_missing`` -- one side carries no usable ``__version``. +* ``unsupported_serialized_dag_schema_version:<n>`` -- ``<n>`` is the first unsupported + version found, base before target. +* ``serialized_dag_canonicalization_failed`` -- a payload could not be normalized (malformed + structure, an unsupported ``client_defaults`` section, an unencodable value). +* ``serialized_dag_recursion_limit_exceeded`` -- the comparison walk ran too deep. +* ``serialized_dag_json_encoding_failed`` -- a value could not be encoded as canonical JSON. + +:meth:`~airflow.models.dag_version.DagVersion.get_diff` reuses +``serialized_dag_decode_failed``, for a stored payload that will not decompress or parse, and +``deadline_alert_missing``, for a payload that references a deadline alert row that is gone. +:func:`build_unavailable_dag_diff` passes its caller's reason through unchanged, so a new +caller extends this list rather than inventing an undocumented value. + +Each change record carries these keys in both disclosure modes: + +* ``path`` -- a JSON-Pointer-style path into a synthetic document with two roots: ``/dag`` + (the canonicalized serialized Dag, with ``__version`` dropped and client defaults folded + into the tasks) and ``/provenance`` (the provenance mapping the caller supplied -- + ``bundle_name``, ``bundle_version`` and ``version_data`` from ``get_diff``). ``~`` and + ``/`` inside a component are escaped as ``~0`` and ``~1``. +* ``operation`` -- ``"added"``, ``"removed"`` or ``"changed"``. +* ``category`` -- a :data:`DiffCategory`: ``task`` for task definitions and the Dag fields + that shape task execution, ``dependency`` for ``dag_dependencies`` and downstream task + ids, ``schedule`` for timetable, dates and concurrency, ``param`` for params, + ``asset`` for task inlets and outlets, ``deadline`` for deadline alerts, + ``callback`` for callback presence, ``metadata`` for descriptive and display fields, + ``authorization`` for ``access_control``, ``provenance`` for everything under + ``/provenance`` plus the file and bundle locators, and ``unknown`` for anything + unclassified. ``_DIFF_V1_DAG_FIELD_CATEGORIES`` is the authoritative Dag-field mapping; + an unclassified task field falls back to ``task``, which over-reports impact. + ``default_args`` is the one Dag field classified per key rather than as a whole, because its + keys are the operator fields they will be applied to: an allowlisted key takes that field's + category (``owner`` is ``metadata``, ``retries`` is ``task``), and the aggregate record for + the rest keeps the conservative ``task``. Only a stored value that is not a readable dict + envelope classifies as ``task`` as a whole. +* ``impact`` -- a :data:`DiffImpact` derived from ``category``: ``provenance``, ``metadata`` + and ``authorization`` carry through unchanged, every operational category becomes + ``execution``, and ``unknown`` stays ``unknown``. +* ``occurrence_count`` -- how many underlying changes the record stands for. + +A caller that authorized values (``include_values=True``, ``values.status`` ``"available"``) +additionally gets: + +* ``before_digest`` / ``after_digest`` -- ``"sha256:<hex>"`` over the canonical JSON of that + side, or ``None`` when that side is missing. Both keys are always present. +* ``before_value`` / ``after_value`` -- the canonicalized value, which is the stored one after + schema and client defaults are folded in and params are normalized. ``before_value`` is absent from + an ``added`` record and ``after_value`` from a ``removed`` one, which is how a missing side + is told apart from a stored ``null``. + +``path`` and ``occurrence_count`` also differ between the modes. A redacted record reports the +public path: members of keyed collections (``/dag/tasks``, ``/dag/dag_dependencies``, +``/dag/tags``, ``/dag/allowed_run_types`` and task group children) collapse to ``*`` instead +of naming a task, tag or group, and records sharing a public path and operation are merged +into one whose ``occurrence_count`` counts the merged changes. An authorized record keeps the +identifying component and always has an ``occurrence_count`` of 1. Fields outside the v1 task, +task group and ``default_args`` allowlists are aggregated into a single ``custom_fields`` record +in both modes, so neither a private serializer field nor a user-chosen ``default_args`` key is +ever named by either -- ``/dag/default_args`` names only allowlisted operator field names, which +is why its keys can be classified at all. A top-level key outside the v1 root allowlist is +likewise reported as ``/custom_fields`` when redacted, while an authorized record keeps its real +root path. + +``max_changes`` bounds the result rather than the traversal: both modes walk the same shape and +reach the same changes, so authorization never decides which changes exist, and the public paths +either mode reports -- and the order of their first occurrence -- are the same. What the bound +admits does differ. An authorized record stands for one change, so the bound caps those directly. +A redacted record stands for every change sharing its public path and operation, and a change +reaching a record already in the result costs no new one, so redacted counting continues past the +bound until a change needs a record the bound will not allow. The same payload at the same bound +can therefore report more occurrences, and a different ``truncated``, when values are withheld. + +``truncated`` reports that such a change was dropped. When it is ``False`` every changed path +is present with an exact ``occurrence_count``. When it is ``True`` the walk stopped early: some +paths are absent rather than unchanged, and every ``occurrence_count`` is a lower bound, since +occurrences after the stop were never reached. + +Typed dictionaries for the result and the change record are deliberately deferred -- the +record shape varies with disclosure mode, so modelling it belongs with the REST layer that +exposes it. :data:`DiffCategory` and :data:`DiffImpact` are stable enough to generate enums +from today. +""" + +from __future__ import annotations + +import copy +import hashlib +import json +from collections.abc import Callable, Mapping, MutableMapping +from contextlib import suppress +from dataclasses import dataclass +from datetime import timedelta +from enum import Enum +from typing import Any, Literal, NamedTuple + +import structlog + +from airflow.serialization.definitions.baseoperator import SerializedBaseOperator +from airflow.serialization.definitions.mappedoperator import SerializedMappedOperator +from airflow.serialization.serialized_objects import ( + _DAG_CALLBACK_FIELDS, + _OPERATOR_TIMEDELTA_FIELDS, + DagSerialization, + OperatorSerialization, +) + +log = structlog.get_logger(__name__) + +DIFF_SCHEMA_VERSION = 1 +DEFAULT_MAX_CHANGES = 500 +MAX_ALLOWED_CHANGES = 5000 +SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS = frozenset((1, 2, 3)) + +DiffCategory = Literal[ + "asset", + "authorization", + "callback", + "deadline", + "dependency", + "metadata", + "param", + "provenance", + "schedule", + "task", + "unknown", +] +DiffImpact = Literal["authorization", "execution", "metadata", "provenance", "unknown"] + +_ORDER_INSENSITIVE_LIST_PATHS = { + ("dag", "tags"), + ("dag", "allowed_run_types"), +} +_KEYED_COLLECTION_PATHS = { + ("dag", "tasks"), + ("dag", "dag_dependencies"), + *_ORDER_INSENSITIVE_LIST_PATHS, +} +# Canonicalization folds away __version and client_defaults and adds provenance, so these are the +# only top-level sections the walk is allowed to name. +_DIFF_V1_PUBLIC_ROOT_FIELDS = frozenset({"dag", "provenance"}) +_CUSTOM_TASK_FIELDS_PATH_COMPONENT = "custom_fields" +# This allowlist is part of diff schema v1. Serializer schema changes must not +# silently change the paths visible to callers of the diff API. +_DIFF_V1_PUBLIC_TASK_FIELDS = frozenset( + { + "__type", + "_can_skip_downstream", + "_disallow_kwargs_override", + "_expand_input_attr", + "_is_empty", + "_is_mapped", + "_is_sensor", + "_logger_name", + "_needs_expansion", + "_operator_extra_links", + "_operator_name", + "_task_display_name", + "_task_module", + "allow_nested_operators", + "depends_on_past", + "do_xcom_push", + "doc", + "doc_json", + "doc_md", + "doc_rst", + "doc_yaml", + "downstream_task_ids", + "email", + "email_on_failure", + "email_on_retry", + "end_date", + "execution_timeout", + "executor", + "executor_config", + "expand_input", + "has_on_execute_callback", + "has_on_failure_callback", + "has_on_retry_callback", + "has_on_skipped_callback", + "has_on_success_callback", + "has_retry_policy", + "ignore_first_depends_on_past", + "inlets", + "is_setup", + "is_stub", + "is_teardown", + "map_index_template", + "max_active_tis_per_dag", + "max_active_tis_per_dagrun", + "max_retry_delay", + "multiple_outputs", + "on_failure_fail_dagrun", + "op_kwargs_expand_input", + "outlets", + "owner", + "params", + "partial_kwargs", + "pool", + "pool_slots", + "priority_weight", + "python_callable_name", + "queue", + "render_template_as_native_obj", + "reschedule", + "resources", + "retries", + "retry_delay", + "retry_exponential_backoff", + "run_as_user", + "start_date", + "start_from_trigger", + "start_trigger_args", + "task_id", + "task_type", + "template_ext", + "template_fields", + "template_fields_renderers", + "trigger_rule", + "ui_color", + "ui_fgcolor", + "wait_for_downstream", + "wait_for_past_depends_before_skipping", + "weight_rule", + } +) +_DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS = frozenset({"_arg_bindings"}) +# Task-field categories used by _get_category. +_DIFF_V1_TASK_ASSET_FIELDS = frozenset({"inlets", "outlets"}) +_DIFF_V1_TASK_PARAM_FIELDS = frozenset({"params"}) +_DIFF_V1_TASK_DEPENDENCY_FIELDS = frozenset({"downstream_task_ids"}) +_DIFF_V1_TASK_METADATA_FIELDS = frozenset( + { + "_operator_name", + "doc", + "doc_json", + "doc_md", + "doc_rst", + "doc_yaml", + "owner", + "ui_color", + "ui_fgcolor", + "_task_display_name", + "task_display_name", + } +) +_DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS = _DIFF_V1_PUBLIC_TASK_FIELDS | {"task_display_name"} +_DEFAULT_ARGS_PATH = ("dag", "default_args") +_RETRY_BACKOFF_FIELD = "retry_exponential_backoff" +# Dag fields the schema gives no default for, added after the first serializer versions. +_ABSENT_AS_NULL_DAG_FIELDS = ("allowed_run_types", "deadline") +# These hydrate to sets (_deserialize_operator_field for a task, TaskGroup.*_ids.update for a group), +# so a reordered stored list means the same edges and must not read as a dependency change. +_SET_VALUED_ID_FIELDS = frozenset( + {"downstream_task_ids", "upstream_task_ids", "downstream_group_ids", "upstream_group_ids"} +) +# A Dag's default_args holds the same operator __init__ kwargs partial_kwargs holds -- the +# serializer rewrites _HAS_FLAG_FIELDS in both -- so it reuses that allowlist. "__type" names the +# stored value's own dict envelope rather than a key inside it. +_DIFF_V1_PUBLIC_DEFAULT_ARGS_FIELDS = _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS - {"__type"} +# Classify every Dag schema field explicitly so new fields require a policy decision. +_DIFF_V1_DAG_FIELD_CATEGORIES: dict[str, DiffCategory] = { + "_concurrency": "schedule", + "_processor_dags_folder": "provenance", + "access_control": "authorization", + "allowed_run_types": "schedule", + "bundle_name": "provenance", + "catchup": "schedule", + "dag_dependencies": "dependency", + "dag_display_name": "metadata", + "dag_id": "metadata", + "dagrun_timeout": "schedule", + "deadline": "deadline", + # Only reached when the stored value is not the dict envelope _collect_default_args_changes + # walks per key; an unreadable envelope keeps the conservative impact of its widest key. + "default_args": "task", + "description": "metadata", + # Decides whether a run pins a bundle version, so it changes which code later runs execute + # and whether triggering a historical version is allowed at all. + "disable_bundle_versioning": "task", + "doc_md": "metadata", + "edge_info": "metadata", + "end_date": "schedule", + "fail_fast": "schedule", + "fileloc": "provenance", + "has_on_failure_callback": "callback", + "has_on_success_callback": "callback", + "is_paused_upon_creation": "schedule", + "max_active_runs": "schedule", + "max_active_tasks": "schedule", + "max_consecutive_failed_dag_runs": "schedule", + "owner_links": "metadata", + "params": "param", + "relative_fileloc": "provenance", + "render_template_as_native_obj": "task", + "rerun_with_latest_version": "task", + "start_date": "schedule", + "tags": "metadata", + "task_group": "task", + "tasks": "task", + "timetable": "schedule", + "timezone": "schedule", +} +# Fields dropped from the current schema that a stored payload can still carry. Every entry must +# survive _canonicalize_payload_v1, or it classifies nothing. +_DIFF_V1_LEGACY_DAG_FIELD_CATEGORIES: dict[str, DiffCategory] = { + "fail_stop": "schedule", + "on_failure_callback": "callback", + "on_success_callback": "callback", + "schedule": "schedule", +} +_RECURSIVE_MAPPING_PATHS = { + (), + ("dag",), + ("provenance",), + *_KEYED_COLLECTION_PATHS, +} +_DIFF_V1_TASK_GROUP_METADATA_FIELDS = frozenset( + {"group_display_name", "tooltip", "doc_md", "ui_color", "ui_fgcolor"} +) +_DIFF_V1_PUBLIC_TASK_GROUP_FIELDS = _DIFF_V1_TASK_GROUP_METADATA_FIELDS | { + "_group_id", + "prefix_group_id", + "children", + "upstream_group_ids", + "downstream_group_ids", + "upstream_task_ids", + "downstream_task_ids", + "expand_input", + "is_mapped", +} + + +def build_unavailable_dag_diff( + *, + base_data: dict[str, Any] | None, + target_data: dict[str, Any] | None, + reason: str, +) -> dict[str, Any]: + """Report a known unavailable reason without comparing the stored payloads.""" + return _mark_unavailable( + _build_diff_result(_get_schema_version(base_data), _get_schema_version(target_data)), reason + ) + + +def build_serialized_dag_diff( + *, + base_data: dict[str, Any] | None, + target_data: dict[str, Any] | None, + base_provenance: Mapping[str, Any] | None = None, + target_provenance: Mapping[str, Any] | None = None, + include_values: bool = False, + max_changes: int = DEFAULT_MAX_CHANGES, +) -> dict[str, Any]: + """ + Compare two serialized Dag payloads and their provenance deterministically. + + Callers must authorize disclosure of the entire serialized payload, including + access-control roles and permissions, before setting ``include_values=True``. + This exposes canonicalized values, digests, and identifying path components. + + ``max_changes`` limits underlying changes admitted to the result. Redacted changes + with the same public path and operation share a record. Repeats can be counted past + the limit until a change needing a new record stops the walk. When ``truncated`` is + true, some paths are absent and occurrence counts are lower bounds. + + An ``unavailable`` result includes the reason comparison failed. + """ + validate_max_changes(max_changes) + + base_schema_version = _get_schema_version(base_data) + target_schema_version = _get_schema_version(target_data) + result = _build_diff_result(base_schema_version, target_schema_version) + + if base_data is None or target_data is None: + return _mark_unavailable(result, "serialized_dag_missing") + + if base_schema_version is None or target_schema_version is None: + return _mark_unavailable(result, "serialized_dag_schema_version_missing") + + unsupported_versions = [ + version + for version in (base_schema_version, target_schema_version) + if version not in SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS + ] + if unsupported_versions: + return _mark_unavailable( + result, f"unsupported_serialized_dag_schema_version:{unsupported_versions[0]}" + ) + + try: + base_document = _canonicalize_payload_v1(base_data) + target_document = _canonicalize_payload_v1(target_data) + base_document["provenance"] = _canonicalize_value(dict(base_provenance or {}), path=("provenance",)) + target_document["provenance"] = _canonicalize_value( + dict(target_provenance or {}), path=("provenance",) + ) + except (AttributeError, KeyError, OverflowError, RecursionError, TypeError, ValueError) as error: + log.warning( + "Serialized Dag diff canonicalization failed", + error_type=type(error).__name__, + base_schema_version=base_schema_version, + target_schema_version=target_schema_version, + ) + return _mark_unavailable(result, "serialized_dag_canonicalization_failed") + + collector = _ChangeCollector(max_changes=max_changes, include_values=include_values) + try: + _collect_changes(base_document, target_document, path=(), collector=collector) + except RecursionError: + log.warning( + "Serialized Dag diff recursion limit exceeded", + base_schema_version=base_schema_version, + target_schema_version=target_schema_version, + ) + return _mark_unavailable(result, "serialized_dag_recursion_limit_exceeded") + except _JsonEncodingError: + log.warning( + "Serialized Dag diff JSON encoding failed", + base_schema_version=base_schema_version, + target_schema_version=target_schema_version, + ) + return _mark_unavailable(result, "serialized_dag_json_encoding_failed") + + result["changes"] = collector.changes + result["truncated"] = collector.is_truncated + if include_values: + result["values"] = {"status": "available"} + return result + + +class _ChangeCollector: + def __init__(self, *, max_changes: int, include_values: bool) -> None: + self.changes: list[dict[str, Any]] = [] + self.count = 0 + self.max_changes = max_changes + self.include_values = include_values + self._truncated = False + self._redacted_changes: dict[tuple[tuple[str, ...], str], dict[str, Any]] = {} + + @property + def is_truncated(self) -> bool: + return self._truncated + + def add( + self, + *, + path: tuple[str, ...], + operation: Literal["added", "removed", "changed"], + before: Any, + after: Any, + ) -> None: + self.count += 1 Review Comment: `self.count += 1` sits above the merge early-return, so a repeat of an already-recorded public path spends budget without producing a record. That makes `max_changes` bound occurrences rather than records, and the redacted default then drops distinct paths. Measured at this HEAD in breeze: 600 tasks all changing `retries` plus one `/dag/timezone` change, `max_changes=500`, redacted, returns a single record (`/dag/tasks/*/retries`, `occurrence_count` 600) with `truncated: true` and no timezone record at all. At 5000 tasks with `max_changes=5000`, which is `MAX_ALLOWED_CHANGES` and the most a caller may ask for, the answer is the same single record, so no permitted bound recovers the timezone change. The same 600-task case at `max_changes=5000` returns both paths with `truncated: false`, which is the shape line 112 promises ("bounds the result rather than the traversal"). I can see this is deliberate: lines 116-119 describe the counting continuing past the bound, and `test_build_diff_excludes_new_group_beyond_change_limit[False-2]` pins it. But it leaves the crowd-out that last round's grouping was asked to fix (#discussion_r4008627247, "the duplicates scale with task count rather than with content") in place, and harder to spot, because the result now reads as small rather than as repetitive. Moving `self.count += 1` below the repeat early-return makes the bound cap records. The redacted walk then stops truncating in practice, but its record count is bounded by the field allowlists rather than by task count, so that is a small set either way. -- 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]
