stijndehaes commented on code in PR #27834:
URL: https://github.com/apache/airflow/pull/27834#discussion_r1031520320
##########
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:
Ok I'll try to make it work in a new commit.
So if I understood it correctly, what is preferred is:
- I keep the old logic that changes the `-1` map_index TI into the `0`
map_index TI
- After that I detect if a `0` map_index ti already exists if so I deleted
the original `-1` TI
--
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]
