stijndehaes commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031541359
##########
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:
I update the code to have this change, please let me know if there is
anything else you would like to see different
--
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]
