https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
TI.task_id == task.task_id,
                TI.run_id == self.run_id,
            )
        )
        existing_indexes = set(query)
        removed_indexes = existing_indexes.difference(range(total_length))
        if removed_indexes:
            session.execute(
                update(TI)
                .where(
                    TI.dag_id == self.dag_id,
                    TI.task_id == task.task_id,
                    TI.run_id == self.run_id,
                    TI.map_index.in_(removed_indexes),
                )
                .values(state=TaskInstanceState.REMOVED)
            )
            session.flush()
        for index in range(total_length):
            if index in existing_indexes:
                continue
            ti = TI(task, run_id=self.run_id, map_index=index, state=None)
            self.log.debug("Expanding TIs upserted %s", ti)
            task_instance_mutation_hook(ti)
            ti = session.merge(ti)
            ti.refresh_from_task(task)
            session.flush()
            yield ti
    @staticmethod
    def get_run(session: Session, dag_id: str, execution_date: datetime) -> DagRun | None: