This is an automated email from the ASF dual-hosted git repository.
potiuk 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 a7ac9379cf9 Read XCom without object reconstruction during dependency
evaluation (#72718)
a7ac9379cf9 is described below
commit a7ac9379cf9d0a443319dd740849c63fff95359c
Author: Jarek Potiuk <[email protected]>
AuthorDate: Mon Oct 19 16:56:52 2026 +0200
Read XCom without object reconstruction during dependency evaluation
(#72718)
* Read XCom without object reconstruction during dependency evaluation
The scheduler evaluates task dependencies in-process, and two of those deps
read
XCom: NotPreviouslySkippedDep reads skipmixin_key for every parent that
inherits
from SkipMixin, and both it and TriggerRuleDep read past_depends_met when
wait_for_past_depends_before_skipping applies.
Both went through TaskInstance.xcom_pull -> XComModel.deserialize_value,
which is
json.loads(value, cls=XComDecoder). XComDecoder's object_hook calls the
Airflow
serialization module with full=True, so any nested dict carrying
__classname__ is
imported and instantiated. The values these deps actually consume are
primitives:
a mapping of lists of task ids, and a boolean. Nothing here needs an object
rebuilt, so the scheduler can read these values as plain JSON.
Add XComModel.deserialize_value_primitives, a plain json.loads with no
object_hook, and LazyXComPrimitivesSelectSequence alongside it. Thread
full=True
through TaskInstance.xcom_pull and pass full=False at the four dep call
sites.
All three of xcom_pull's return paths are covered -- the direct one, and
both
lazy-sequence ones, which is the path a mapped task instance takes. The skip
decisions of SkipMixin tasks in the same mapped task group, which
NotPreviouslySkippedDep reads with one query per pass, are decoded the same
way.
NotPreviouslySkippedDep also now checks that the skipmixin payload is a
mapping
before testing membership in it. It reads whatever the parent task wrote,
so a
scalar there previously raised TypeError inside dependency evaluation
rather than
being ignored. Because the payload is now plain JSON, a set or tuple of
task ids
arrives as its serde wrapper; its items are read from the wrapper without
constructing anything, and a decision whose task ids are not a list of
strings is
ignored the same way.
Task code pulling another task's return value is unchanged and still gets
the
original Python object; only the dependency-evaluation reads are narrowed.
Generated-by: Claude Opus 5
* Treat only a literal True as past-depends-met during dependency evaluation
Generated-by: Claude Opus 5 following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Ignore skip decisions with a non-string classname
Generated-by: Claude Opus 5
---
airflow-core/src/airflow/models/taskinstance.py | 19 +-
airflow-core/src/airflow/models/xcom.py | 43 +++
.../ti_deps/deps/not_previously_skipped_dep.py | 66 ++++-
.../src/airflow/ti_deps/deps/prev_dagrun_dep.py | 14 +
.../src/airflow/ti_deps/deps/trigger_rule_dep.py | 12 +-
airflow-core/tests/unit/models/test_xcom.py | 90 ++++++
.../deps/test_not_previously_skipped_dep.py | 311 +++++++++++++++++++++
.../unit/ti_deps/deps/test_trigger_rule_dep.py | 62 +++-
.../src/tests_common/test_utils/mock_context.py | 1 +
9 files changed, 597 insertions(+), 21 deletions(-)
diff --git a/airflow-core/src/airflow/models/taskinstance.py
b/airflow-core/src/airflow/models/taskinstance.py
index 4a0ed099426..f569fc4f36f 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/airflow-core/src/airflow/models/taskinstance.py
@@ -118,7 +118,7 @@ from airflow.models.hitl import HITLDetail # noqa: F401
from airflow.models.log import Log
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.models.taskreschedule import TaskReschedule
-from airflow.models.xcom import XCOM_RETURN_KEY, LazyXComSelectSequence
+from airflow.models.xcom import XCOM_RETURN_KEY,
LazyXComPrimitivesSelectSequence, LazyXComSelectSequence
from airflow.serialization.enums import stringify_encoding_keys
from airflow.settings import task_instance_mutation_hook
from airflow.task.priority_strategy import
validate_and_load_priority_weight_strategy
@@ -2839,9 +2839,16 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
map_indexes: int | Iterable[int] | None = None,
default: Any = None,
run_id: str | None = None,
+ full: bool = True,
) -> Any:
""":meta private:""" # noqa: D400
# This is only kept for compatibility in tests for now while AIP-72 is
in progress.
+ #
+ # `full=False` returns plain JSON types and never runs the Airflow
+ # serialization module, so a stored `__classname__` envelope is not
imported or
+ # instantiated. Callers that evaluate task dependencies -- the
ti_deps, run by
+ # the scheduler and by the API server's task instance dependencies
endpoint --
+ # pass it, since they only need primitive values.
from airflow.models.xcom import XComModel, xcom_entity
@@ -2879,9 +2886,12 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
return default
if map_indexes is not None or first.map_index < 0:
- return XComModel.deserialize_value(first)
+ if full:
+ return XComModel.deserialize_value(first)
+ return XComModel.deserialize_value_primitives(first)
- return LazyXComSelectSequence.from_select(
+ sequence_cls = LazyXComSelectSequence if full else
LazyXComPrimitivesSelectSequence
+ return sequence_cls.from_select(
query.with_only_columns(entity.value).order_by(None),
order_by=[entity.map_index.expression],
session=session,
@@ -2907,7 +2917,8 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
ordering.append(case(map_index_whens, value=entity.map_index))
else:
ordering.append(entity.map_index)
- return LazyXComSelectSequence.from_select(
+ sequence_cls = LazyXComSelectSequence if full else
LazyXComPrimitivesSelectSequence
+ return sequence_cls.from_select(
query.with_only_columns(entity.value).order_by(None),
order_by=ordering,
session=session,
diff --git a/airflow-core/src/airflow/models/xcom.py
b/airflow-core/src/airflow/models/xcom.py
index 2054ba1368a..b86ade0fa5e 100644
--- a/airflow-core/src/airflow/models/xcom.py
+++ b/airflow-core/src/airflow/models/xcom.py
@@ -456,6 +456,34 @@ class _XComOperations:
# Already deserialized (e.g., set via Task Execution API)
return result.value
+ @staticmethod
+ def deserialize_value_primitives(result: Any) -> Any:
+ """
+ Deserialize an XCom value to plain JSON types only.
+
+ Unlike :meth:`deserialize_value` this never runs the Airflow
serialization
+ module, so a ``__classname__`` envelope in the stored value is
returned as an
+ ordinary ``dict`` instead of being imported and instantiated.
+
+ This is what dependency evaluation uses, in the scheduler and in the
API
+ server's task instance dependencies endpoint. The values it consumes
are
+ primitives, so XCom is read without reconstructing objects in that
process. Callers that genuinely need the
+ original Python object (task code pulling another task's return value)
keep
+ using :meth:`deserialize_value`.
+
+ :param result: The XCom database row or object containing a ``value``
attribute.
+ :return: The value as plain JSON types (``dict``, ``list``, ``str``,
``int``,
+ ``float``, ``bool`` or ``None``).
+ """
+ if result.value is None:
+ return None
+
+ try:
+ return json.loads(result.value)
+ except (ValueError, TypeError):
+ # Already deserialized (e.g., set via Task Execution API)
+ return result.value
+
def _rows():
from airflow.models.dagrun import DagRun
@@ -687,6 +715,21 @@ class LazyXComSelectSequence(LazySelectSequence[Any]):
return get_xcom_model().deserialize_value(row)
+class LazyXComPrimitivesSelectSequence(LazyXComSelectSequence):
+ """
+ Like :class:`LazyXComSelectSequence`, but never reconstructs Python
objects.
+
+ Used by XCom reads during dependency evaluation. See
+ :meth:`XComModel.deserialize_value_primitives`.
+
+ :meta private:
+ """
+
+ @staticmethod
+ def _process_row(row: Row) -> Any:
+ return get_xcom_model().deserialize_value_primitives(row)
+
+
__compat_imports = {
"BaseXCom": "airflow.sdk.bases.xcom",
"XCom": "airflow.sdk.execution_time.xcom",
diff --git
a/airflow-core/src/airflow/ti_deps/deps/not_previously_skipped_dep.py
b/airflow-core/src/airflow/ti_deps/deps/not_previously_skipped_dep.py
index d0bef5f7665..80c3380f877 100644
--- a/airflow-core/src/airflow/ti_deps/deps/not_previously_skipped_dep.py
+++ b/airflow-core/src/airflow/ti_deps/deps/not_previously_skipped_dep.py
@@ -17,11 +17,12 @@
# under the License.
from __future__ import annotations
+import logging
from typing import TYPE_CHECKING
-from airflow.models.taskinstance import PAST_DEPENDS_MET
from airflow.models.xcom import XComModel, xcom_entity
from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
+from airflow.ti_deps.deps.prev_dagrun_dep import is_past_depends_met
from airflow.utils.state import TaskInstanceState
if TYPE_CHECKING:
@@ -34,6 +35,8 @@ if TYPE_CHECKING:
from airflow.ti_deps.dep_context import DepContext
from airflow.ti_deps.deps.base_ti_dep import TIDepStatus
+log = logging.getLogger(__name__)
+
# The following constants are taken from the SkipMixin class in the standard
provider
# The key used by SkipMixin to store XCom data.
XCOM_SKIPMIXIN_KEY = "skipmixin_key"
@@ -89,18 +92,25 @@ class NotPreviouslySkippedDep(BaseTIDep):
if dep_context.has_regions(ti, session=session)
else ti.region_index
)
+ # `full=False`: the payload is a plain mapping of lists of
task ids, so
+ # the scheduler only needs primitive values here and reads
XCom without
+ # reconstructing objects.
prev_result = ti.xcom_pull(
task_ids=parent.task_id,
key=XCOM_SKIPMIXIN_KEY,
session=session,
map_indexes=xcom_map_index,
+ full=False,
)
if prev_result is None:
# This can happen if the parent task has not yet run.
continue
- if _should_skip(prev_result, ti.task_id,
is_direct_parent=True):
+ if (decision := _skip_decision(prev_result, parent.task_id))
is None:
+ continue
+
+ if _should_skip(decision, ti.task_id, is_direct_parent=True):
yield from self._skip(ti, dep_context, parent.task_id,
session=session)
return
@@ -131,10 +141,7 @@ class NotPreviouslySkippedDep(BaseTIDep):
# ti should be skipped, set ti.state to SKIPPED and fail the rule so
that the
# ti does not execute.
if dep_context.wait_for_past_depends_before_skipping:
- past_depends_met = ti.xcom_pull(
- task_ids=ti.task_id, key=PAST_DEPENDS_MET, session=session,
default=False
- )
- if not past_depends_met:
+ if not is_past_depends_met(ti, session=session):
yield self._failing_status(reason="Task should be skipped but
the past depends are not met")
return
ti.set_state(TaskInstanceState.SKIPPED, session=session)
@@ -143,6 +150,49 @@ class NotPreviouslySkippedDep(BaseTIDep):
)
+# Collection types serde writes as ``{"__classname__": ..., "__version__":
..., "__data__": [...]}``.
+_SERDE_COLLECTION_CLASSNAMES = frozenset(
+ {"builtins.list", "builtins.tuple", "builtins.set", "builtins.frozenset"}
+)
+
+
+def _task_id_list(value: object) -> list[str] | None:
+ # The decision is read as plain JSON, so a set or tuple of task ids
arrives as the serde
+ # wrapper dict; take its items without importing or constructing anything.
+ if isinstance(value, dict):
+ classname = value.get("__classname__")
+ if not (isinstance(classname, str) and classname in
_SERDE_COLLECTION_CLASSNAMES):
+ return None
+ value = value.get("__data__")
+ if isinstance(value, (list, tuple)) and all(isinstance(task_id, str) for
task_id in value):
+ return list(value)
+ return None
+
+
+def _skip_decision(prev_result: object, parent_task_id: str) -> dict[str,
list[str]] | None:
+ """
+ Return the SkipMixin decision with its task id collections as lists, or
``None`` if malformed.
+
+ A task may write anything under this key, so a value of another shape is
treated as absent
+ rather than failing the dep.
+ """
+ if not isinstance(prev_result, dict):
+ problem = f"expected a mapping, got {type(prev_result).__name__}"
+ else:
+ decision = {}
+ for field in (XCOM_SKIPMIXIN_FOLLOWED, XCOM_SKIPMIXIN_SKIPPED):
+ if field not in prev_result:
+ continue
+ if (task_ids := _task_id_list(prev_result[field])) is None:
+ problem = f"{field!r} is not a collection of task ids"
+ break
+ decision[field] = task_ids
+ else:
+ return decision
+ log.warning("Ignoring malformed %s XCom from task %s: %s",
XCOM_SKIPMIXIN_KEY, parent_task_id, problem)
+ return None
+
+
def _should_skip(prev_result: dict, task_id: str, *, is_direct_parent: bool)
-> bool:
# "followed" only lists a branch operator's direct downstream tasks, so it
says nothing
# about a task further down, which follows its own trigger rule instead.
@@ -190,7 +240,9 @@ def _mapped_group_skip_decisions(
for row in rows:
if (state := finished_states.get((row.task_id, row.map_index))) is
None:
continue
- if (decision := XComModel.deserialize_value(row)) is not None:
+ # Primitives only, as for a direct parent's decision above.
+ stored = XComModel.deserialize_value_primitives(row)
+ if stored is not None and (decision := _skip_decision(stored,
row.task_id)) is not None:
decisions.setdefault(row.map_index, []).append(
(row.task_id, state == TaskInstanceState.SUCCESS, decision)
)
diff --git a/airflow-core/src/airflow/ti_deps/deps/prev_dagrun_dep.py
b/airflow-core/src/airflow/ti_deps/deps/prev_dagrun_dep.py
index 056c0c667b8..1762abb49cd 100644
--- a/airflow-core/src/airflow/ti_deps/deps/prev_dagrun_dep.py
+++ b/airflow-core/src/airflow/ti_deps/deps/prev_dagrun_dep.py
@@ -24,6 +24,7 @@ from sqlalchemy import func, or_, select
from airflow.models.backfill import BackfillDagRun
from airflow.models.dagrun import DagRun
from airflow.models.taskinstance import PAST_DEPENDS_MET, TaskInstance as TI
+from airflow.models.xcom import LazyXComSelectSequence
from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
from airflow.utils.db import exists_query
from airflow.utils.session import provide_session
@@ -37,6 +38,19 @@ if TYPE_CHECKING:
_SUCCESSFUL_STATES = (TaskInstanceState.SKIPPED, TaskInstanceState.SUCCESS)
+def is_past_depends_met(ti: TI, *, session: Session) -> bool:
+ """Return whether this task instance recorded that its past depends were
met."""
+ value = ti.xcom_pull(
+ task_ids=ti.task_id, key=PAST_DEPENDS_MET, session=session,
default=False, full=False
+ )
+ # XCom is read as plain JSON here, so only the literal ``True`` this dep
pushes counts; a
+ # non-empty mapping or other truthy value does not. A mapped task
instance's read returns
+ # the value of every map index.
+ if isinstance(value, LazyXComSelectSequence):
+ return any(item is True for item in value)
+ return value is True
+
+
class PrevDagrunDep(BaseTIDep):
"""
Is the past dagrun in a state that allows this task instance to run.
diff --git a/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py
b/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py
index 3b52d4db366..9ccc4adcb13 100644
--- a/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py
+++ b/airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py
@@ -26,9 +26,9 @@ from uuid import UUID
from sqlalchemy import and_, func, or_, select
-from airflow.models.taskinstance import PAST_DEPENDS_MET
from airflow.task.trigger_rule import TriggerRule as TR
from airflow.ti_deps.deps.base_ti_dep import BaseTIDep
+from airflow.ti_deps.deps.prev_dagrun_dep import is_past_depends_met
from airflow.utils.state import TaskInstanceState
if TYPE_CHECKING:
@@ -370,10 +370,7 @@ class TriggerRuleDep(BaseTIDep):
new_state == TaskInstanceState.SKIPPED
and dep_context.wait_for_past_depends_before_skipping
):
- past_depends_met = ti.xcom_pull(
- task_ids=ti.task_id, key=PAST_DEPENDS_MET,
session=session, default=False
- )
- if not past_depends_met:
+ if not is_past_depends_met(ti, session=session):
yield (
self._failing_status(
reason="Task should be skipped but the past
depends are not met"
@@ -531,10 +528,7 @@ class TriggerRuleDep(BaseTIDep):
new_state == TaskInstanceState.SKIPPED
and dep_context.wait_for_past_depends_before_skipping
):
- past_depends_met = ti.xcom_pull(
- task_ids=ti.task_id, key=PAST_DEPENDS_MET,
session=session, default=False
- )
- if not past_depends_met:
+ if not is_past_depends_met(ti, session=session):
yield self._failing_status(
reason="Task should be skipped but the past
depends are not met"
)
diff --git a/airflow-core/tests/unit/models/test_xcom.py
b/airflow-core/tests/unit/models/test_xcom.py
index 422985e6c9e..1a413aabfb0 100644
--- a/airflow-core/tests/unit/models/test_xcom.py
+++ b/airflow-core/tests/unit/models/test_xcom.py
@@ -587,3 +587,93 @@ class TestXComRoundTrip:
deserialized_value = XComModel.deserialize_value(mock_xcom)
assert deserialized_value == expected_value
+
+
+_TUPLE_ENVELOPE = {"__classname__": "builtins.tuple", "__version__": 1,
"__data__": [1, 2]}
+
+
[email protected]
+def no_serde_deserialize():
+ with mock.patch("airflow.sdk.serde.deserialize",
side_effect=AssertionError("must not construct")) as m:
+ yield m
+
+
+class TestXComPrimitives:
+ @pytest.mark.parametrize(
+ ("value", "expected"),
+ [
+ pytest.param(json.dumps(_TUPLE_ENVELOPE), _TUPLE_ENVELOPE,
id="classname-str"),
+ pytest.param(json.dumps({"a": [_TUPLE_ENVELOPE]}), {"a":
[_TUPLE_ENVELOPE]}, id="nested"),
+ pytest.param(_TUPLE_ENVELOPE, _TUPLE_ENVELOPE,
id="already-loaded"),
+ pytest.param(None, None, id="none"),
+ ],
+ )
+ def test_deserialize_value_primitives(self, value, expected,
no_serde_deserialize):
+ result = XComModel.deserialize_value_primitives(MagicMock(value=value))
+
+ assert result == expected
+ no_serde_deserialize.assert_not_called()
+
+ @pytest.fixture
+ def mapped_tis(self, session, task_instance):
+ tis = []
+ for map_index in (0, 1):
+ ti = create_task_instance(
+ EmptyOperator(task_id="mapped"),
+ run_id=task_instance.run_id,
+ dag_version_id=task_instance.dag_version_id,
+ map_index=map_index,
+ )
+ ti.dag_id = task_instance.dag_id
+ session.add(ti)
+ tis.append(ti)
+ session.commit()
+ return tis
+
+ @staticmethod
+ def _set_envelope(session, ti):
+ XComModel.set(
+ key="xcom_1",
+ value=json.dumps(_TUPLE_ENVELOPE),
+ serialize=False,
+ dag_id=ti.dag_id,
+ task_id=ti.task_id,
+ run_id=ti.run_id,
+ map_index=ti.map_index,
+ session=session,
+ )
+
+ def test_xcom_pull_full_reconstructs(self, session, task_instance):
+ self._set_envelope(session, task_instance)
+
+ assert task_instance.xcom_pull(task_ids="task_1", key="xcom_1",
session=session) == (1, 2)
+
+ def test_xcom_pull_primitives_single_value(self, session, task_instance,
no_serde_deserialize):
+ self._set_envelope(session, task_instance)
+
+ result = task_instance.xcom_pull(task_ids="task_1", key="xcom_1",
session=session, full=False)
+
+ assert result == _TUPLE_ENVELOPE
+ no_serde_deserialize.assert_not_called()
+
+ def test_xcom_pull_primitives_multiple_tasks(self, session,
task_instances, no_serde_deserialize):
+ for ti in task_instances:
+ self._set_envelope(session, ti)
+
+ result = task_instances[0].xcom_pull(
+ task_ids=["task_1", "task_2"], key="xcom_1", session=session,
full=False
+ )
+
+ assert type(result).__name__ == "LazyXComPrimitivesSelectSequence"
+ assert list(result) == [_TUPLE_ENVELOPE, _TUPLE_ENVELOPE]
+ no_serde_deserialize.assert_not_called()
+
+ def test_xcom_pull_primitives_mapped_task(self, session, task_instance,
mapped_tis, no_serde_deserialize):
+ for ti in mapped_tis:
+ self._set_envelope(session, ti)
+
+ result = task_instance.xcom_pull(task_ids="mapped", key="xcom_1",
session=session, full=False)
+
+ assert type(result).__name__ == "LazyXComPrimitivesSelectSequence"
+ assert list(result) == [_TUPLE_ENVELOPE, _TUPLE_ENVELOPE]
+ no_serde_deserialize.assert_not_called()
diff --git
a/airflow-core/tests/unit/ti_deps/deps/test_not_previously_skipped_dep.py
b/airflow-core/tests/unit/ti_deps/deps/test_not_previously_skipped_dep.py
index b5b79f62fb7..0c0fa53b524 100644
--- a/airflow-core/tests/unit/ti_deps/deps/test_not_previously_skipped_dep.py
+++ b/airflow-core/tests/unit/ti_deps/deps/test_not_previously_skipped_dep.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import json
from unittest import mock
import pendulum
@@ -24,6 +25,7 @@ import pytest
from sqlalchemy import delete, select
from airflow.models import DagRun, TaskInstance
+from airflow.models.taskinstance import PAST_DEPENDS_MET
from airflow.models.xcom import XComModel
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import BranchPythonOperator
@@ -36,6 +38,7 @@ from airflow.ti_deps.deps.not_previously_skipped_dep import (
XCOM_SKIPMIXIN_SKIPPED,
NotPreviouslySkippedDep,
)
+from airflow.ti_deps.deps.prev_dagrun_dep import is_past_depends_met
from airflow.utils.state import State
from airflow.utils.types import DagRunType
@@ -653,3 +656,311 @@ def
test_branch_skip_decision_bypasses_custom_xcom_backend(session, dag_maker):
dep = NotPreviouslySkippedDep()
assert len(list(dep.get_dep_statuses(tis["op2"], DepContext(),
session=session))) == 1
assert tis["op2"].state == State.SKIPPED
+
+
+def test_skipmixin_xcom_with_classname_is_not_reconstructed(session,
dag_maker):
+ """
+ A serde envelope stored under the skipmixin key is read as a plain mapping.
+
+ The Execution API stores the caller's value verbatim. The scheduler only
needs
+ primitive values to evaluate this dep, so XCom is read without importing a
class
+ or reconstructing an object.
+ """
+ start_date = pendulum.datetime(2020, 1, 1)
+ with dag_maker(
+ "test_skipmixin_xcom_with_classname_dag",
+ schedule=None,
+ start_date=start_date,
+ session=session,
+ ):
+ op1 = BranchPythonOperator(task_id="op1", python_callable=lambda:
"op2")
+ op2 = EmptyOperator(task_id="op2")
+ op1 >> op2
+
+ dagrun = dag_maker.create_dagrun(run_type=DagRunType.MANUAL,
state=State.RUNNING)
+ ti, ti2 = dagrun.task_instances
+ run_task_instance(ti, op1)
+
+ # Overwrite the parent's skipmixin XCom with a serialization envelope, the
shape
+ # `XComDecoder.object_hook` would hand to `deserialize(..., full=True)`.
+ # `serialize=False` with a pre-serialized string is exactly what the
Execution API
+ # does with a caller-supplied value -- it stores the task's bytes verbatim.
+ XComModel.set(
+ key=XCOM_SKIPMIXIN_KEY,
+ value='{"__classname__": "builtins.dict", "__version__": 1,
"__data__": {}}',
+ serialize=False,
+ task_id=ti.task_id,
+ dag_id=ti.dag_id,
+ run_id=ti.run_id,
+ map_index=ti.map_index,
+ session=session,
+ )
+ session.commit()
+
+ dep = NotPreviouslySkippedDep()
+
+ # The envelope is read as an inert mapping. It carries neither "followed"
nor
+ # "skipped", so it yields no skip decision and the dep passes.
+ assert len(list(dep.get_dep_statuses(ti2, DepContext(), session=session)))
== 0
+ assert dep.is_met(ti2, session=session)
+ assert ti2.state != State.SKIPPED
+
+
+def test_non_mapping_skipmixin_xcom_is_ignored(session, dag_maker):
+ """
+ A skipmixin XCom that is not a mapping is ignored rather than crashing the
scheduler.
+
+ Without the shape check the membership test runs against whatever the task
wrote, so
+ a scalar raises TypeError inside dependency evaluation.
+ """
+ start_date = pendulum.datetime(2020, 1, 1)
+ with dag_maker(
+ "test_non_mapping_skipmixin_xcom_dag",
+ schedule=None,
+ start_date=start_date,
+ session=session,
+ ):
+ op1 = BranchPythonOperator(task_id="op1", python_callable=lambda:
"op2")
+ op2 = EmptyOperator(task_id="op2")
+ op1 >> op2
+
+ dagrun = dag_maker.create_dagrun(run_type=DagRunType.MANUAL,
state=State.RUNNING)
+ ti, ti2 = dagrun.task_instances
+ run_task_instance(ti, op1)
+
+ XComModel.set(
+ key=XCOM_SKIPMIXIN_KEY,
+ value="5", # a bare scalar: `"followed" in 5` raises TypeError
+ serialize=False,
+ task_id=ti.task_id,
+ dag_id=ti.dag_id,
+ run_id=ti.run_id,
+ map_index=ti.map_index,
+ session=session,
+ )
+ session.commit()
+
+ dep = NotPreviouslySkippedDep()
+ assert len(list(dep.get_dep_statuses(ti2, DepContext(), session=session)))
== 0
+ assert ti2.state != State.SKIPPED
+
+
[email protected](
+ "stored",
+ [
+ pytest.param('{"__classname__": "builtins.dict", "__version__": 1,
"__data__": {}}', id="classname"),
+ pytest.param("5", id="scalar"),
+ ],
+)
+def test_mapped_task_group_skip_decision_read_as_primitives(session,
dag_maker, stored):
+ """A decision of a SkipMixin task in the same mapped task group is read
the same way as a direct parent's."""
+ with dag_maker("test_mapped_group_primitives_dag", schedule=None,
session=session):
+
+ @task.short_circuit(task_id="gate")
+ def gate(value):
+ return value
+
+ @task_group
+ def group(value):
+ gate(value) >> EmptyOperator(task_id="child")
+
+ group.expand(value=[True, False])
+
+ dr, tis = _create_run(dag_maker)
+ _finish_with_skip_decisions(
+ dr, tis, "group.gate", {0: {XCOM_SKIPMIXIN_SKIPPED: ["group.child"]}},
session=session
+ )
+ XComModel.set(
+ key=XCOM_SKIPMIXIN_KEY,
+ value=stored,
+ serialize=False,
+ dag_id=dr.dag_id,
+ task_id="group.gate",
+ run_id=dr.run_id,
+ map_index=1,
+ session=session,
+ )
+ session.flush()
+
+ dep = NotPreviouslySkippedDep()
+
+ assert not dep.is_met(tis[("group.child", 0)], session=session)
+ assert tis[("group.child", 0)].state == State.SKIPPED
+ assert dep.is_met(tis[("group.child", 1)], session=session)
+ assert tis[("group.child", 1)].state != State.SKIPPED
+
+
+def _serde_collection(classname, items):
+ return {"__classname__": classname, "__version__": 1, "__data__":
list(items)}
+
+
[email protected](
+ ("decision", "serialize", "expected_skipped"),
+ [
+ pytest.param({XCOM_SKIPMIXIN_FOLLOWED: {"op3"}}, True, {"op2"},
id="followed-set"),
+ pytest.param({XCOM_SKIPMIXIN_SKIPPED: frozenset({"op2"})}, True,
{"op2"}, id="skipped-frozenset"),
+ pytest.param(
+ json.dumps({XCOM_SKIPMIXIN_FOLLOWED:
_serde_collection("builtins.tuple", ["op3"])}),
+ False,
+ {"op2"},
+ id="followed-tuple",
+ ),
+ pytest.param(
+ json.dumps({XCOM_SKIPMIXIN_SKIPPED:
_serde_collection("builtins.list", ["op3"])}),
+ False,
+ {"op3"},
+ id="skipped-list",
+ ),
+ pytest.param({XCOM_SKIPMIXIN_FOLLOWED: ["op2", "op3"]}, True, set(),
id="followed-list"),
+ ],
+)
+def test_skipmixin_xcom_collection_wrapper_is_read_as_task_ids(
+ session, dag_maker, decision, serialize, expected_skipped
+):
+ """A set or tuple of task ids, stored as a serde wrapper, is read as its
items without constructing it."""
+ with dag_maker("test_skipmixin_collection_wrapper_dag", schedule=None,
session=session):
+ op1 = BranchPythonOperator(task_id="op1", python_callable=lambda:
"op3")
+ op1 >> [EmptyOperator(task_id="op2"), EmptyOperator(task_id="op3")]
+
+ dr = dag_maker.create_dagrun(run_type=DagRunType.MANUAL,
state=State.RUNNING)
+ tis = {ti.task_id: ti for ti in dr.task_instances}
+ tis["op1"].state = State.SUCCESS
+ session.merge(tis["op1"])
+ XComModel.set(
+ key=XCOM_SKIPMIXIN_KEY,
+ value=decision,
+ serialize=serialize,
+ dag_id=dr.dag_id,
+ task_id="op1",
+ run_id=dr.run_id,
+ session=session,
+ )
+ session.flush()
+
+ dep = NotPreviouslySkippedDep()
+ with mock.patch("airflow.sdk.serde.deserialize",
side_effect=AssertionError("must not construct")):
+ for task_id in ("op2", "op3"):
+ assert dep.is_met(tis[task_id], session=session) is (task_id not
in expected_skipped)
+ assert {task_id for task_id in ("op2", "op3") if tis[task_id].state ==
State.SKIPPED} == expected_skipped
+
+
[email protected](
+ "stored",
+ [
+ pytest.param({XCOM_SKIPMIXIN_FOLLOWED: "op3"}, id="followed-str"),
+ pytest.param({XCOM_SKIPMIXIN_SKIPPED: [1]}, id="skipped-non-str"),
+ pytest.param(
+ {XCOM_SKIPMIXIN_FOLLOWED: {"__classname__": "builtins.set",
"__version__": 1, "__data__": "op3"}},
+ id="wrapper-data-not-list",
+ ),
+ pytest.param(
+ {XCOM_SKIPMIXIN_FOLLOWED:
_serde_collection("collections.OrderedDict", ["op3"])},
+ id="wrapper-other-class",
+ ),
+ pytest.param({XCOM_SKIPMIXIN_FOLLOWED: {"__classname__": []}},
id="wrapper-classname-list"),
+ pytest.param({XCOM_SKIPMIXIN_SKIPPED: {"__classname__": {}}},
id="wrapper-classname-dict"),
+ ],
+)
+def test_malformed_skipmixin_task_id_collection_is_ignored(session, dag_maker,
stored):
+ """A decision whose task id collection has another shape is ignored, so no
task is skipped by it."""
+ with dag_maker("test_malformed_task_id_collection_dag", schedule=None,
session=session):
+ op1 = BranchPythonOperator(task_id="op1", python_callable=lambda:
"op3")
+ op1 >> [EmptyOperator(task_id="op2"), EmptyOperator(task_id="op3")]
+
+ dr = dag_maker.create_dagrun(run_type=DagRunType.MANUAL,
state=State.RUNNING)
+ tis = {ti.task_id: ti for ti in dr.task_instances}
+ tis["op1"].state = State.SUCCESS
+ session.merge(tis["op1"])
+ XComModel.set(
+ key=XCOM_SKIPMIXIN_KEY,
+ value=json.dumps(stored),
+ serialize=False,
+ dag_id=dr.dag_id,
+ task_id="op1",
+ run_id=dr.run_id,
+ session=session,
+ )
+ session.flush()
+
+ dep = NotPreviouslySkippedDep()
+ for task_id in ("op2", "op3"):
+ assert dep.is_met(tis[task_id], session=session)
+ assert tis[task_id].state != State.SKIPPED
+
+
+_PAST_DEPENDS_MET_VALUES = [
+ pytest.param(True, True, True, id="true"),
+ pytest.param(False, True, False, id="false"),
+ pytest.param(
+ '{"__classname__": "builtins.tuple", "__version__": 1, "__data__":
[1]}', False, False, id="envelope"
+ ),
+]
+
+
[email protected](("stored", "serialize", "met"),
_PAST_DEPENDS_MET_VALUES)
+def test_wait_for_past_depends_before_skipping_reads_stored_value(session,
dag_maker, stored, serialize, met):
+ """Only a stored literal ``True`` lets the task be skipped, and the value
is never reconstructed."""
+ with dag_maker("test_past_depends_met_dag", schedule=None,
session=session):
+ op1 = BranchPythonOperator(task_id="op1", python_callable=lambda:
"op3")
+ op1 >> [EmptyOperator(task_id="op2"), EmptyOperator(task_id="op3")]
+
+ dr = dag_maker.create_dagrun(run_type=DagRunType.MANUAL,
state=State.RUNNING)
+ tis = {ti.task_id: ti for ti in dr.task_instances}
+ tis["op1"].state = State.SUCCESS
+ session.merge(tis["op1"])
+ for task_id, key, value, serialize_value in (
+ ("op1", XCOM_SKIPMIXIN_KEY, {XCOM_SKIPMIXIN_FOLLOWED: ["op3"]}, True),
+ ("op2", PAST_DEPENDS_MET, stored, serialize),
+ ):
+ XComModel.set(
+ key=key,
+ value=value,
+ serialize=serialize_value,
+ dag_id=dr.dag_id,
+ task_id=task_id,
+ run_id=dr.run_id,
+ session=session,
+ )
+ session.flush()
+
+ dep_context = DepContext(wait_for_past_depends_before_skipping=True)
+ with mock.patch("airflow.sdk.serde.deserialize",
side_effect=AssertionError("must not construct")) as m:
+ statuses = list(NotPreviouslySkippedDep().get_dep_statuses(tis["op2"],
dep_context, session=session))
+ m.assert_not_called()
+
+ assert len(statuses) == 1
+ assert not statuses[0].passed
+ if met:
+ assert tis["op2"].state == State.SKIPPED
+ else:
+ assert "past depends are not met" in statuses[0].reason
+ assert tis["op2"].state != State.SKIPPED
+
+
[email protected](("stored", "serialize", "met"),
_PAST_DEPENDS_MET_VALUES)
+def test_is_past_depends_met_for_mapped_task_instance(session, dag_maker,
stored, serialize, met):
+ """A mapped task instance's read returns a sequence; it counts only a
literal ``True`` in it."""
+ with dag_maker("test_past_depends_met_mapped_dag", schedule=None,
session=session):
+
+ @task_group
+ def group(value):
+ EmptyOperator(task_id="child")
+
+ group.expand(value=[1, 2])
+
+ dr, tis = _create_run(dag_maker)
+ child = tis[("group.child", 0)]
+ XComModel.set(
+ key=PAST_DEPENDS_MET,
+ value=stored,
+ serialize=serialize,
+ dag_id=dr.dag_id,
+ task_id=child.task_id,
+ run_id=dr.run_id,
+ map_index=child.map_index,
+ session=session,
+ )
+ session.flush()
+
+ with mock.patch("airflow.sdk.serde.deserialize",
side_effect=AssertionError("must not construct")):
+ assert is_past_depends_met(child, session=session) is met
diff --git a/airflow-core/tests/unit/ti_deps/deps/test_trigger_rule_dep.py
b/airflow-core/tests/unit/ti_deps/deps/test_trigger_rule_dep.py
index 53d33165d0e..27fb20d1b78 100644
--- a/airflow-core/tests/unit/ti_deps/deps/test_trigger_rule_dep.py
+++ b/airflow-core/tests/unit/ti_deps/deps/test_trigger_rule_dep.py
@@ -31,7 +31,8 @@ from sqlalchemy import delete, event
import airflow.settings
from airflow.models.dag_version import DagVersion
from airflow.models.dynamic_region import SENTINEL_REGION_ID, DynamicRegion
-from airflow.models.taskinstance import TaskInstance
+from airflow.models.taskinstance import PAST_DEPENDS_MET, TaskInstance
+from airflow.models.xcom import XComModel
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import task, task_group
@@ -2295,6 +2296,65 @@ def
test_setup_constraint_wait_for_past_depends_before_skipping(
)
+_PAST_DEPENDS_MET_VALUES = [
+ pytest.param(True, True, True, id="true"),
+ pytest.param(False, True, False, id="false"),
+ pytest.param(
+ '{"__classname__": "builtins.tuple", "__version__": 1, "__data__":
[1]}', False, False, id="envelope"
+ ),
+ pytest.param("1", False, False, id="truthy-int"),
+]
+
+
[email protected](("stored", "serialize", "met"),
_PAST_DEPENDS_MET_VALUES)
[email protected](
+ ("task_kwargs", "skipped_reason"),
+ [
+ pytest.param(
+ {"trigger_rule": TriggerRule.ALL_SUCCESS, "normal_tasks":
["FakeTaskID", "OtherFakeTaskID"]},
+ "requires all upstream tasks to have succeeded, but found 1",
+ id="trigger-rule",
+ ),
+ pytest.param(
+ {"trigger_rule": TriggerRule.ALL_DONE, "setup_tasks":
["FakeTaskID", "OtherFakeTaskID"]},
+ "",
+ id="setup-constraint",
+ ),
+ ],
+)
+def test_wait_for_past_depends_before_skipping_reads_stored_value(
+ session, get_task_instance, task_kwargs, skipped_reason, stored,
serialize, met
+):
+ """Only a stored literal ``True`` lets the task be skipped, and the value
is never reconstructed."""
+ ti = get_task_instance(
+ success=1, skipped=1, failed=0, removed=0, upstream_failed=0, done=2,
**task_kwargs
+ )
+ XComModel.set(
+ key=PAST_DEPENDS_MET,
+ value=stored,
+ serialize=serialize,
+ dag_id=ti.dag_id,
+ task_id=ti.task_id,
+ run_id=ti.run_id,
+ map_index=ti.map_index,
+ session=session,
+ )
+ session.flush()
+
+ with mock.patch("airflow.sdk.serde.deserialize",
side_effect=AssertionError("must not construct")) as m:
+ _test_trigger_rule(
+ ti=ti,
+ session=session,
+ flag_upstream_failed=True,
+ wait_for_past_depends_before_skipping=True,
+ expected_ti_state=SKIPPED if met else None,
+ expected_reason=(
+ skipped_reason if met else "Task should be skipped but the
past depends are not met"
+ ),
+ )
+ m.assert_not_called()
+
+
@pytest.mark.parametrize(("flag_upstream_failed", "expected_ti_state"),
[(True, SKIPPED), (False, None)])
@pytest.mark.need_serialized_dag
def test_setup_mapped_task_group_finished_upstream_before_expand(
diff --git a/devel-common/src/tests_common/test_utils/mock_context.py
b/devel-common/src/tests_common/test_utils/mock_context.py
index 4e7aa7884f2..70c5f8c6729 100644
--- a/devel-common/src/tests_common/test_utils/mock_context.py
+++ b/devel-common/src/tests_common/test_utils/mock_context.py
@@ -51,6 +51,7 @@ def mock_context(task) -> Context:
map_indexes: int | Iterable[int] | None = None,
default: Any = None,
run_id: str | None = None,
+ full: bool = True,
) -> Any:
key = f"{self.task_id}_{self.dag_id}_{key}"
if map_indexes is not None and (not isinstance(map_indexes, int)
or map_indexes >= 0):
