uranusjr commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031172476
##########
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:
The fact that the original task instance is _converted_ instead of deleted
was explicitly mentioned in the original proposal, so it would be best if we
could keep the behaviour (I think there are some subtle differences but don’t
recall exactly what). Would it be possible?
--
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]
