steveahnahn commented on code in PR #72668:
URL: https://github.com/apache/airflow/pull/72668#discussion_r3981472455
##########
airflow-core/src/airflow/ti_deps/deps/trigger_rule_dep.py:
##########
@@ -89,6 +89,28 @@ def calculate(cls, finished_upstreams:
Iterator[TaskInstance]) -> _UpstreamTISta
)
+def _covers_expansion_one_to_one(task, counted_task_ids: Collection[str]) ->
bool:
+ """
+ Whether the counted upstream instances map 1:1 onto this task's expansion.
+
+ Inferring instance removal from upstream state counts is only sound when
this task expands
+ over exactly one XCom reference, that reference is the entire counted
upstream set, and no
+ literal or mapped task group multiplies the length. Otherwise the task's
expansion length
+ has no relation to the counts: comparing ``map_index`` against the success
count would
+ remove valid instances (static lists, zips, multiple upstreams).
+ """
+ if task.get_closest_mapped_task_group() is not None:
+ return False
+ iter_deps = getattr(task, "iter_mapped_dependencies", None)
+ if iter_deps is None:
+ return False
+ referenced = [op.task_id for op in iter_deps()]
+ if len(referenced) != 1 or set(referenced) != set(counted_task_ids):
+ return False
+ expand_value = task._get_specified_expand_input().value
+ return not isinstance(expand_value, Mapping) or len(expand_value) == 1
Review Comment:
Good catch, the gate now requires the referenced upstream to be expanded.
Pushed in 37b2ea82e4
--
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]
