uranusjr commented on code in PR #27506:
URL: https://github.com/apache/airflow/pull/27506#discussion_r1016216761
##########
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:
Yes but it still makes sure the task is _ready_ to be expanded (via
`TriggerRuleDep`). I don’t think the mini scheduler does this; it simply tries
to make downstreams until it succeeds.
Although looking at the implementation, `expand_mapped_task` will fail a
task if expansion is attempted before it’s ready, so BackfillJob probably
already does _something_ to make sure this does not happen (although I’m not
sure what).
Let’s change this to WARNING (or maybe ERROR is better since it’s really not
supposed to happen) and see what happens.
--
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]
