https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
"""Try to expand the ti, if needed.
            If the ti needs expansion, newly created task instances are
            returned as well as the original ti.
            The original ti is also modified in-place and assigned the
            ``map_index`` of 0.
            If the ti does not need expansion, either because the task is not
            mapped, or has already been expanded, *None* is returned.
            """
            if ti.map_index >= 0:  
                return None
            from airflow.models.mappedoperator import MappedOperator
            if isinstance(ti.task, MappedOperator):
                
                
                
                ti.clear_db_references(session=session)
            try:
                expanded_tis, _ = ti.task.expand_mapped_task(self.run_id, session=session)
            except NotMapped:  
                return None
            if expanded_tis:
                return expanded_tis
            return ()
        
        expansion_happened = False
        
        revised_map_index_task_ids = set()
        for schedulable in itertools.chain(schedulable_tis, additional_tis):