uranusjr commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031173795
##########
airflow/models/abstractoperator.py:
##########
@@ -494,18 +493,15 @@ def expand_mapped_task(self, run_id: str, *, session:
Session) -> tuple[Sequence
total_length,
)
unmapped_ti.state = TaskInstanceState.SKIPPED
- indexes_to_map = ()
else:
- # Otherwise convert this into the first mapped index, and
create
- # TaskInstance for other indexes.
- unmapped_ti.map_index = 0
- self.log.debug("Updated in place to become %s", unmapped_ti)
- all_expanded_tis.append(unmapped_ti)
- indexes_to_map = range(1, total_length)
- state = unmapped_ti.state
- elif not total_length:
+ # Delete the original task so we can create new ones
+ self.log.debug("Deleting the original task instance: %s",
unmapped_ti)
+ state = unmapped_ti.state
+ session.delete(unmapped_ti)
Review Comment:
If the scheduler already reached a faulty state (e.g. `[-1, 0, 1]` in your
case) we can of course delete the ti to rescure it, but otherwise it’d be best
to not delete things.
--
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]
