args, dag: DAG, ti: TaskInstance | TaskInstancePydantic
) -> None | TaskReturnCode:
    """
    Run the task based on a mode.
    Any of the 3 modes are available:
    - using LocalTaskJob
    - as raw task
    - by executor
    """
    assert not isinstance(ti, TaskInstancePydantic), "Wait for AIP-44 implementation to complete"
    if args.local:
        return _run_task_by_local_task_job(args, ti)
    if args.raw:
        return _run_raw_task(args, ti)
    _run_task_by_executor(args, dag, ti)
    return None
def _run_task_by_executor(args, dag: DAG, ti: TaskInstance) -> None:
    """
    Send the task to the executor for execution.
    This can result in the task being started by another host if the executor implementation does.
    """
    pickle_id = None
    if args.ship_dag: