dabla commented on code in PR #62922:
URL: https://github.com/apache/airflow/pull/62922#discussion_r2891127404
##########
task-sdk/src/airflow/sdk/definitions/mappedoperator.py:
##########
@@ -336,19 +364,20 @@ def __repr__(self):
return f"<Mapped({self.task_type}): {self.task_id}>"
def __attrs_post_init__(self):
- from airflow.sdk.definitions.xcom_arg import XComArg
-
- if self.get_closest_mapped_task_group() is not None:
- raise NotImplementedError("operator expansion in an expanded task
group is not yet supported")
-
- if self.task_group:
- self.task_group.add(self)
- if self.dag:
- self.dag.add_task(self)
- XComArg.apply_upstream_relationship(self,
self._get_specified_expand_input().value)
- for k, v in self.partial_kwargs.items():
- if k in self.template_fields:
- XComArg.apply_upstream_relationship(self, v)
+ if self._apply_upstream_relationship:
+ from airflow.sdk.definitions.xcom_arg import XComArg
+
+ if self.get_closest_mapped_task_group() is not None:
+ raise NotImplementedError("operator expansion in an expanded
task group is not yet supported")
+
+ if self.task_group:
+ self.task_group.add(self)
+ if self.dag:
Review Comment:
This is normal obviously an inherent to the solution, the iteration or
looping of the operator is done in memory through mapped runtime TI's which in
fact don't really exist nor in Scheduler or Dag itself as they are purely there
to make the execution of the iterated operator possible, hence why it's never
added otherwise it would raise a DuplicateTaskIdFound. For Airflow it is as if
only one operator (not mapped operators) is being executed but in a loop with
different arguments.
--
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]
