Add start execution from triggerer support to dynamic task mapping

[Lee-W]

This is a follow-up PR of #39585, which enables the mapped operator to start execution from triggerer

Why

Dynamic task mapping is not yet supported in #38674 and #39585.

What

Allow DAG authors to provide argument trigger_kwargs and start_from_trigger in __init__ which can be used in expand.

from __future__ import annotations

from datetime import timedelta
from typing import Any

from airflow.models.baseoperator import BaseOperator, StartTriggerArgs
from airflow.triggers.temporal import TimeDeltaTrigger


class AsyncOperator(BaseOperator):
    start_trigger_args = StartTriggerArgs(
        trigger_cls="airflow.triggers.testing.SuccessTrigger",
        trigger_kwargs={},
        next_method="execute_complete",
        timeout=None,
    )
    start_from_trigger = True

    def __init__(self, *args, start_from_trigger: bool, trigger_kwargs=None, **kwargs):
        super().__init__(*args, **kwargs)
        self.start_from_trigger = start_from_trigger
        self.start_trigger_args.trigger_kwargs = trigger_kwargs

    def execute(self, context):
        self.log.info("execute")

    def execute_complete(self, context, event=None) -> None:
        self.log.info("execute complete")

from __future__ import annotations

import datetime

import pendulum

from airflow import DAG
from airflow.operators.async_op import AsyncOperator

with DAG(
    dag_id="example_async_operator",
    schedule=None,
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    dagrun_timeout=datetime.timedelta(minutes=10),
) as dag:
    add_one_task = AsyncOperator.partial(task_id="add_one").expand(
        trigger_kwargs=[{}, {}], start_from_trigger=[True, False]
    )

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[Lee-W]

The code here is not yet well structured but has been tested locally and seems to work fine. It would be great if I could get some reviews and suggestions. Thanks!

[Lee-W]

This PR is ready to be reviewed but blocked by #39585 so I'll keep it as draft for now.

[phanikumv]

lgtm

[Lee-W]

@uranusjr @jedcunningham I plan on merging this one later today or early tomorrow. Please let me know if there's still concern or if you want to look deeper. Thanks!

[uranusjr]

I was thinking how we could get rid of the very out-of-place isinstance calls. The main problem is we don’t need context for the non-mapped case. Looking into the implementation though, the resolution process really only needs ti and run_id, and getting both are basically free at those parts of the code. So I would probably do

1. This this PR, create an interface in AbstractOperator (say expand_start_from_trigger) that takes context and session. The BaseOperator implementation will simply return self.start_from_trigger.
2. If this is fine performance-wise, just call it a day.
3. Assuming we want to optimise, create a new PR to only pass in {"run_id": self.run_id, "ti": ti} instead with a comment explaining why (performance consideration, and reasons this should be enough).

[uranusjr]

Probably shouldn’t be an approval (the exception class likely needs discussion, and I would prefer the function in AbstractOperator to be a part of this PR), but close enough.

[Lee-W]

The latest comments from @uranusjr have been addressed. I'll go ahead and merge this one.