SameerMesiah97 commented on code in PR #59691:
URL: https://github.com/apache/airflow/pull/59691#discussion_r3269784728
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -2367,6 +2375,54 @@ def _visit_relevant_relatives_for_mapped(mapped_tasks:
Iterable[tuple[str, int]]
return visited
+def _resolve_placeholder_map_index(
+ *,
+ task: Operator,
+ relative: Operator,
+ map_index: int,
+ run_id: str,
+ session: Session,
+) -> int | None:
+ """
+ Resolve the correct map_index for upstream dependency evaluation.
+
+ This handles the transition from map_index = -1 (pre-expansion placeholder)
+ to map_index = 0 (post-expansion placeholder successor).
+
+ Returns:
+ - 0 if the placeholder has transitioned from -1 to 0
+ - None if no override should be applied
+ """
+ if map_index != -1:
+ return None
+
+ rows = session.execute(
+ select(TaskInstance.task_id, TaskInstance.map_index).where(
+ TaskInstance.dag_id == relative.dag_id,
+ TaskInstance.run_id == run_id,
+ TaskInstance.task_id.in_([task.task_id, relative.task_id]),
+ TaskInstance.map_index.in_([-1, 0]),
+ )
+ ).all()
+
+ task_to_map_indexes: dict[str, list[int]] = defaultdict(list)
+ for task_id, mi in rows:
+ task_to_map_indexes[task_id].append(mi)
+
+ # We only rewrite when:
+ # 1) the current task is still using the placeholder (-1)
+ # 2) the upstream placeholder (-1) no longer exists
+ # 3) the post-expansion placeholder (0) does exist
+ if (
+ -1 in task_to_map_indexes.get(task.task_id, [])
+ and -1 not in task_to_map_indexes.get(relative.task_id, [])
+ and 0 in task_to_map_indexes.get(relative.task_id, [])
+ ):
Review Comment:
The dictionary `task_to_map_indexes` is now being accessed directly.
--
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]
