https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
*, task_instance: TaskInstance | TaskInstancePydantic, task: Operator, value: Any, session: Session
) -> None:
    """
    Record the task map for downstream tasks.
    :param task_instance: the task instance
    :param task: The task object
    :param value: The value
    :param session: SQLAlchemy ORM Session
    :meta private:
    """
    if next(task.iter_mapped_dependants(), None) is None:  
        return
    
    
    
    
    if isinstance(task, MappedOperator):
        return
    if value is None:
        raise XComForMappingNotPushed()
    if not _is_mappable_value(value):
        raise UnmappableXComTypePushed(value)
    task_map = TaskMap.from_task_instance_xcom(task_instance, value)
    max_map_length = conf.getint("core", "max_map_length", fallback=1024)
    if task_map.length > max_map_length:
        raise UnmappableXComLengthPushed(value, max_map_length)
    session.merge(task_map)
def _get_previous_dagrun(