https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
another_task(inp, val)  # val here should resolve to [2, 4, 6].
            tg2.expand(inp=["a", "b"])
        The surrounding mapped task groups of ``upstream`` and ``self.task`` are
        inspected to find a common "ancestor". If such an ancestor is found,
        we need to return specific map indexes to pull a partial value from
        upstream XCom.
        :param upstream: The referenced upstream task.
        :param ti_count: The total count of task instance this task was expanded
            by the scheduler, i.e. ``expanded_ti_count`` in the template context.
        :return: Specific map index or map indexes to pull, or ``None`` if we
            want to "whole" return value (i.e. no mapped task groups involved).
        """
        
        
        
        
        
        if not ti_count:
            return None
        
        
        
        
        common_ancestor = _find_common_ancestor_mapped_group(self.task, upstream)
        if common_ancestor is None:
            return None
        
        
        
        ancestor_ti_count = common_ancestor.get_mapped_ti_count(self.run_id, session=session)