uranusjr commented on code in PR #27506:
URL: https://github.com/apache/airflow/pull/27506#discussion_r1016007433
##########
airflow/models/mappedoperator.py:
##########
@@ -620,13 +620,19 @@ def expand_mapped_task(self, run_id: str, *, session:
Session) -> tuple[Sequence
try:
total_length =
self._get_specified_expand_input().get_total_map_length(run_id, session=session)
except NotFullyPopulated as e:
- self.log.info(
- "Cannot expand %r for run %s; missing upstream values: %s",
- self,
- run_id,
- sorted(e.missing),
- )
- total_length = None
+ if self.dag and self.dag.partial:
+ # partial dags comes from the mini scheduler. It's
+ # likely that the upstream tasks are not yet done,
+ # so we ignore this exception.
+ total_length = None
+ else:
+ self.log.info(
Review Comment:
This can be changed to WARNING if we are sure it’s only supposed to be
called when the task is ready for expansion (i.e. all referenced upstreams have
finished).
The DagRun implementation ensures this by ensuring expansion only happens
(line 755) when a task’s dependencies are all met (line 746):
https://github.com/apache/airflow/blob/98a9c574e3e30b3aea710c23e18955e88871c9e6/airflow/models/dagrun.py#L744-L766
but I don’t quite recall if the backfill job does the same. Can someone
track this down please?
--
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]
