steveahnahn commented on code in PR #72668:
URL: https://github.com/apache/airflow/pull/72668#discussion_r3981469385
##########
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
Review Comment:
Done, uses is_mapped() and task: Operator now
--
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]
