You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dstandish (via GitHub)" <gi...@apache.org> on 2025/04/30 17:23:48 UTC
[PR] Use latest bundle version when clearing / re-running dag [airflow]
dstandish opened a new pull request, #50040:
URL: https://github.com/apache/airflow/pull/50040
(no comment)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #50040:
URL: https://github.com/apache/airflow/pull/50040#discussion_r2075479358
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -285,7 +285,12 @@ def clear_task_instances(
# the task is terminated and becomes eligible for retry.
ti.state = TaskInstanceState.RESTARTING
else:
- ti_dag = dag if dag and dag.dag_id == ti.dag_id else dag_bag.get_dag(ti.dag_id, session=session)
+ dr = ti.dag_run
Review Comment:
Is it guaranteed that ti.dag_run cannot be none?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50040:
URL: https://github.com/apache/airflow/pull/50040#discussion_r2075484766
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -285,7 +285,12 @@ def clear_task_instances(
# the task is terminated and becomes eligible for retry.
ti.state = TaskInstanceState.RESTARTING
else:
- ti_dag = dag if dag and dag.dag_id == ti.dag_id else dag_bag.get_dag(ti.dag_id, session=session)
+ dr = ti.dag_run
Review Comment:
i believe this guarantees it?
<img width="498" alt="image" src="https://github.com/user-attachments/assets/f5d7b27c-ba3a-4431-ad22-39c221bffac5" />
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #50040:
URL: https://github.com/apache/airflow/pull/50040#issuecomment-2854862595
Btw `clear_task_instances` can't be used in tasks in 3.0+ anyway since it uses DB
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #50040:
URL: https://github.com/apache/airflow/pull/50040#discussion_r2075494680
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -285,7 +285,12 @@ def clear_task_instances(
# the task is terminated and becomes eligible for retry.
ti.state = TaskInstanceState.RESTARTING
else:
- ti_dag = dag if dag and dag.dag_id == ti.dag_id else dag_bag.get_dag(ti.dag_id, session=session)
+ dr = ti.dag_run
Review Comment:
Ok. I think here: https://github.com/apache/airflow/blob/468bfe339ed3d5053db2a9776b6f9bbf4f7a04de/airflow-core/src/airflow/models/dag.py#L1048
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50040:
URL: https://github.com/apache/airflow/pull/50040#discussion_r2075431703
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -326,6 +327,19 @@ def clear_task_instances(
if dr.state in State.finished_dr_states:
dr.state = dag_run_state
dr.start_date = timezone.utcnow()
+ if TYPE_CHECKING:
+ assert dag # todo: change signature so this is required
+ if not dag.disable_bundle_versioning:
+ if dr.dag_model:
+ bundle_version = dr.dag_model.bundle_version
+ else:
+ bundle_version = session.scalar(
+ select(DagModel.bundle_version).where(
+ DagModel.dag_id == dag.dag_id,
+ )
+ )
Review Comment:
There were some scenarios where the dag run object did not have a dag model --- I don't recall exactly --- but it probably came out in some tests
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50040:
URL: https://github.com/apache/airflow/pull/50040#discussion_r2071346225
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -326,6 +327,19 @@ def clear_task_instances(
if dr.state in State.finished_dr_states:
dr.state = dag_run_state
dr.start_date = timezone.utcnow()
+ if TYPE_CHECKING:
+ assert dag # todo: change signature so this is required
+ if not dag.disable_bundle_versioning:
+ if dr.dag_model:
+ bundle_version = dr.dag_model.bundle_version
+ else:
+ bundle_version = session.scalar(
+ select(DagModel.bundle_version).where(
+ DagModel.dag_id == dag.dag_id,
+ )
+ )
Review Comment:
When is this useful? Some kind of db state handling issue?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #50040:
URL: https://github.com/apache/airflow/pull/50040#issuecomment-2854663082
> This looks good. My worry is that the clear_task_instances was not a private function. Is it ok to change it to a private function now?
Right -- understandable.
My thinking is that it was not intentional to make that a public function, and that I am "fixing" that in this patch release.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on PR #50040:
URL: https://github.com/apache/airflow/pull/50040#issuecomment-2854729843
> > This looks good. My worry is that the clear_task_instances was not a private function. Is it ok to change it to a private function now?
>
> Right -- understandable.
>
> My thinking is that it was not intentional to make that a public function, and that I am "fixing" that in this patch release.
The function has been around for a while(Airflow 2 or 1 not sure) and it looks like people use it in their DAGs e.g https://github.com/apache/airflow/discussions/30237 and https://github.com/apache/airflow/issues/19329#issuecomment-1269642000
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr merged PR #50040:
URL: https://github.com/apache/airflow/pull/50040
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #50040:
URL: https://github.com/apache/airflow/pull/50040#discussion_r2071350440
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -326,6 +327,19 @@ def clear_task_instances(
if dr.state in State.finished_dr_states:
dr.state = dag_run_state
dr.start_date = timezone.utcnow()
+ if TYPE_CHECKING:
+ assert dag # todo: change signature so this is required
Review Comment:
Since this function has a DagBag object, I’d just auto-populate the variable instead. This is actually done above on line 288 for another situation.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50040:
URL: https://github.com/apache/airflow/pull/50040#discussion_r2075434167
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -326,6 +327,19 @@ def clear_task_instances(
if dr.state in State.finished_dr_states:
dr.state = dag_run_state
dr.start_date = timezone.utcnow()
+ if TYPE_CHECKING:
+ assert dag # todo: change signature so this is required
Review Comment:
i changed the signature.
there's not much point in passing dag in as a param to this function since the TIs can be from different dags.
moreover, they can be from different serdags associated with the same dag -- and that in particular is what makes this complicated and required me to go and use the scheduler dag bag object
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Use latest bundle version when clearing / re-running dag [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on code in PR #50040:
URL: https://github.com/apache/airflow/pull/50040#discussion_r2075936423
##########
airflow-core/src/airflow/models/taskinstance.py:
##########
@@ -326,6 +327,19 @@ def clear_task_instances(
if dr.state in State.finished_dr_states:
dr.state = dag_run_state
dr.start_date = timezone.utcnow()
+ if TYPE_CHECKING:
+ assert dag # todo: change signature so this is required
+ if not dag.disable_bundle_versioning:
+ if dr.dag_model:
+ bundle_version = dr.dag_model.bundle_version
+ else:
+ bundle_version = session.scalar(
+ select(DagModel.bundle_version).where(
+ DagModel.dag_id == dag.dag_id,
+ )
+ )
Review Comment:
removed this logic
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org