ashb commented on a change in pull request #21614:
URL: https://github.com/apache/airflow/pull/21614#discussion_r811225561
##########
File path: airflow/models/dagrun.py
##########
@@ -649,27 +651,46 @@ def task_instance_scheduling_decisions(self, session:
Session = NEW_SESSION) ->
def _get_ready_tis(
self,
- scheduleable_tasks: List[TI],
+ schedulable_tis: List[TI],
finished_tis: List[TI],
session: Session,
) -> Tuple[List[TI], bool]:
old_states = {}
ready_tis: List[TI] = []
changed_tis = False
- if not scheduleable_tasks:
+ if not schedulable_tis:
return ready_tis, changed_tis
+ # If we expand TIs, we need a new list so that we iterate over them
too. (We can't alter
+ # `schedulable_tis` in place and have the `for` loop pick them up
+ expanded_tis: List[TI] = []
+
# Check dependencies
- for st in scheduleable_tasks:
- old_state = st.state
- if st.are_dependencies_met(
+ for schedulable in itertools.chain(schedulable_tis, expanded_tis):
+
+ # Expansion of last resort! This is ideally handled in the
mini-scheduler in LocalTaskJob, but if
+ # for any reason it wasn't, we need to expand it now
+ if schedulable.map_index < 0 and schedulable.task.is_mapped:
Review comment:
I think leave this for now, and we can optimize/benchmark later
--
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]
