xBis7 commented on code in PR #71737:
URL: https://github.com/apache/airflow/pull/71737#discussion_r4312642796
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1479,8 +1517,27 @@ def recalculate(self) -> _UnfinishedStates:
@provide_session
def task_instance_scheduling_decisions(self, *, session: Session =
NEW_SESSION) -> TISchedulingDecision:
- tis = self.get_task_instances(session=session, state=State.task_states)
- self.log.debug("number of tis tasks for %s: %s task(s)", self,
len(tis))
+ # Fetch finished task instances as lightweight read-only rows and only
the
+ # unfinished ones as full ORM objects. Finished ones are never mutated
during
+ # scheduling, so hydrating them is pure overhead.
+ #
+ # Flush here to avoid task instances ending up in the wrong list or in
both,
+ # in case of a pending in-session state change.
+ # The airflow session maker sets autoflush=False (ref
settings.configure_orm).
+ session.flush()
+ unfinished_tis = self.get_task_instances(session=session,
state=State.unfinished)
+ finished_rows = session.execute(
+ select(*FINISHED_TI_COLUMNS)
+ .where(
+ TI.dag_id == self.dag_id,
+ TI.run_id == self.run_id,
+ TI.state.in_(State.finished),
Review Comment:
Done
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1479,8 +1517,27 @@ def recalculate(self) -> _UnfinishedStates:
@provide_session
def task_instance_scheduling_decisions(self, *, session: Session =
NEW_SESSION) -> TISchedulingDecision:
- tis = self.get_task_instances(session=session, state=State.task_states)
- self.log.debug("number of tis tasks for %s: %s task(s)", self,
len(tis))
+ # Fetch finished task instances as lightweight read-only rows and only
the
+ # unfinished ones as full ORM objects. Finished ones are never mutated
during
+ # scheduling, so hydrating them is pure overhead.
+ #
+ # Flush here to avoid task instances ending up in the wrong list or in
both,
+ # in case of a pending in-session state change.
+ # The airflow session maker sets autoflush=False (ref
settings.configure_orm).
+ session.flush()
+ unfinished_tis = self.get_task_instances(session=session,
state=State.unfinished)
+ finished_rows = session.execute(
+ select(*FINISHED_TI_COLUMNS)
+ .where(
+ TI.dag_id == self.dag_id,
+ TI.run_id == self.run_id,
+ TI.state.in_(State.finished),
+ )
+ .order_by(TI.task_id, TI.map_index)
Review Comment:
Done
--
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]
