Add dynamic task mapping into TaskSDK runtime

[ashb]

The big change here (other than just moving code around) is to introduce a
conceptual separation between Definition/Execution time and Scheduler time.

This means that the expansion of tasks (creating the TaskInstance rows with
different map_index values) is now done on the scheduler, and we now
deserialize to different classes. For example, when we deserialize the
DictOfListsExpandInput it gets turned into an instance of
SchedulerDictOfListsExpandInput. This is primarily designed so that DB access
is kept 100% out of the TaskSDK.

Some of the changes here are on the "wat" side of the scale, and this is
mostly designed to not break 100% of our tests, and we have #45549 to look at
that more holistically.

To support the "reduce" style task which takes as input a sequence of all the
pushed (mapped) XCom values, and to keep the previous behaviour of not loading
all values in to memory at once, we have added a new HEAD route to the Task
Execution interface that returns the number of mapped XCom values so that it
is possible to implement __len__ on the new LazyXComSequence class.

I have deleted a tranche of tests from tests/models that were to do with
runtime behavoir and and now tested in the TaskSDK instead.

Testing:

I have been testing this with Celery Executor and the example_dynamic_task_mapping dag, which is this:

with DAG(dag_id="example_dynamic_task_mapping", schedule=None, start_date=datetime(2022, 3, 4)) as dag:
    @task
    def add_one(x: int):
        return x + 1

    @task
    def sum_it(values):
        total = sum(values)
        print(f"Total was {total}")

    added_values = add_one.expand(x=[1, 2, 3])
    sum_it(added_values)

It now works, and the sum_it tasks logs show:

{"timestamp":"2025-02-04T14:24:05.264679Z","level":"info","event":"Total was 9","chan":"stdout","logger":"task"}

Closes #44360

---

^ 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.

[uranusjr]

This looks good to me. I’d be even more confident if the map_index changes are taken out, but I’ll take Ash’s words they don’t affect anything else if they’re left in. Aside from that it’s just the seemingly redundant type() calls left. We should remove them if possible, or find out why they are needed.

[ashb]

This looks good to me. I’d be even more confident if the map_index changes are taken out, but I’ll take Ash’s words they don’t affect anything else if they’re left in. Aside from that it’s just the seemingly redundant type() calls left. We should remove them if possible, or find out why they are needed.

Yeah, I'll back out the map index change and try/remove the type as it shouldn't be needed. Just sorting out all the typing and static checks right now first.

[jedcunningham]

Overall LGTM.

[ashb]

Use the command line to resolve conflicts before continuing.
providers/src/airflow/providers/cncf/kubernetes/operators/pod.py

This is a "fake" conflict. git on the CLI does it without complaining (it just got renamed). If this current run passes I am merging after rebasing without waiting for next tests to pass.