Add user-facing docs for the new Task Loops feature.

Task Loops, introduced in AIP-111, are a whole new way for dag authors to thing
Loops chaNge how a Dag author thinks about repeating work, and the change to
implement it reaches storage, the scheduler, the Execution API and the UI. Since
all these changes are relatively large, it is helpful before we introduct those
that reviewers have a mental model of what the behaviour should be, and the
features we want to allow are. Rather than explaining this just in PR messages
or review threads, lets add this as user facing docs, as most of that will be
needed there too.

The new Loops doc covers how a loop decides whether to run another pass, how
a task reads the previous pass's result, what clearing part of a loop does to
the rest, and how loops interact with mapped tasks. At this point/in this PR,
nothing exists as code mind you.

The short "loop or mapped tasks" page is net-new, and is ther mostly as a
jumping off point for "here are the dynamic features you can build your dag
with".

This oVerview is also why "dynamic mapped tasks" is getting a name change.
Mapping is no longer the only way for a Dag's shape to depend on data at
runtime, and arguably, mapped tasks aren't really that "dynamic" anymore.
"Mapped tasks" is what the UI and the API already call the result ("Clear All
Mapped Tasks", MappedOperator), and it reads next to "Loops" as the title of the
page the overview links to. "Task mapping" never appears in the UI.

However, please note that the rename is "soft" on purpose. The mappIng page
keeps its file name and its `dynamic-task-mapping` label, and the old name stays
in its opening sentence so existing searches still land. Provider docs link that
page by path through intersphinx so moving the file would break their builds
until each one is updated. The `dynamic-task-mapping` capability key that the
language SDKs report is an identifier rather than prose and is left alone.

The snippets are inline here because the example Dags that the guide will
include ship with the authoring API, and a file that calls `.loop()` cannot be
imported before `.loop()` exists. That change replaces them with included copies
that a test executes, so the documented code cannot drift from the behaviour.

The placeholder newsfragment is also here because `.loop()` is a new
user-visible Dag authoring feature that ships through task-sdk with
airflow-core. It will grow and be adjusted by future PRs in this stack.

Author: ashb

airflow-core/docs/administration-and-deployment/dag-serialization.rst

@@ -187,7 +187,7 @@ The system automatically fills in any missing values by walking up the hierarchy
 MappedOperator Default Handling
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-MappedOperators (dynamic task mapping) also participate in the default value system:
+MappedOperators (task mapping) also participate in the default value system:
 
 .. code-block:: python
 

airflow-core/docs/authoring-and-scheduling/deferring.rst

@@ -395,7 +395,7 @@ After the trigger has finished executing, the task may be sent back to the worke
             start_from_trigger: bool,
             **kwargs: dict[str, Any],
         ) -> None:
-            # This whole method will be skipped during dynamic task mapping.
+            # This whole method will be skipped during task mapping.
 
             super().__init__(*args, **kwargs)
             self.start_trigger_args.trigger_kwargs = trigger_kwargs

airflow-core/docs/authoring-and-scheduling/dynamic-task-mapping.rst

@@ -16,25 +16,29 @@
     under the License.
 
 .. _dynamic-task-mapping:
+.. _mapped-tasks:
 
-====================
-Dynamic Task Mapping
-====================
+============
+Mapped tasks
+============
 
-Dynamic Task Mapping allows a way for a workflow to create a number of tasks at runtime based upon current data, rather than the Dag author having to know in advance how many tasks would be needed.
+Task mapping, also called dynamic task mapping, allows a way for a workflow to create a number of tasks at runtime based upon current data, rather than the Dag author having to know in advance how many tasks would be needed.
 
 This is similar to defining your tasks in a for loop, but instead of having the DAG file fetch the data and do that itself, the scheduler can do this based on the output of an upstream task.
-Unlike a Python for-loop executed at DAG parse time, dynamic task mapping defers task creation until runtime, allowing the scheduler to determine the exact number of task instances based on upstream task outputs.
+Unlike a Python for-loop executed at DAG parse time, task mapping defers task creation until runtime, allowing the scheduler to determine the exact number of task instances based on upstream task outputs.
 Right before a mapped task is executed the scheduler will create *n* copies of the task, one for each input.
 
 It is also possible to have a task operate on the collected output of a mapped task, commonly known as map and reduce.
 
+.. seealso::
+   Not sure whether you need mapped tasks or a loop? See :ref:`loops-and-mapped-tasks`.
+
 Simple mapping
 ==============
 
 In its simplest form you can map over a list defined directly in your Dag file using the ``expand()`` function instead of calling your task directly.
 
-If you want to see a simple usage of Dynamic Task Mapping, you can look below:
+If you want to see a simple usage of task mapping, you can look below:
 
 .. exampleinclude:: /../src/airflow/example_dags/example_dynamic_task_mapping.py
     :language: python
@@ -388,7 +392,7 @@ For example, this code will *not* work:
 
 When code in ``my_task_group`` is executed, ``value`` would still only be a reference, not the real value, so the ``if not value`` branch will not work as you likely want. However, if you pass that reference into a task, it will become resolved when the task is executed, and the three ``my_task`` instances will therefore receive 1, 2, and 3, respectively.
 
-It is, therefore, important to remember that, if you intend to perform any logic on a value passed into a task group function, you must always use a task to run the logic, such as ``@task.branch`` (or ``BranchPythonOperator``) for conditions, and task mapping methods for loops.
+It is, therefore, important to remember that, if you intend to perform any logic on a value passed into a task group function, you must always use a task to run the logic, such as ``@task.branch`` (or ``BranchPythonOperator``) for conditions, and ``expand()`` to iterate over values. To repeat a group of tasks, see :ref:`loops-and-mapped-tasks`.
 
 .. note:: Task-mapping in a mapped task group is not permitted
 

airflow-core/docs/authoring-and-scheduling/examples/example_task_loops.py

@@ -0,0 +1,114 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Repeat task groups with fixed counts, runtime conditions, and mapped tasks."""
+
+from __future__ import annotations
+
+# [START refine_estimate]
+from airflow.sdk import dag, task, task_group
+
+
+@dag(schedule=None, catchup=False, tags=["example"])
+def refine_estimate():
+    @task_group
+    def refine():
+        @task
+        def improve(*, loop):
+            estimate = 1.0 if loop.index == 0 else loop.previous["estimate"]
+            return (estimate + 2.0 / estimate) / 2.0
+
+        @task
+        def evaluate(estimate):
+            return {"estimate": estimate, "error": abs(estimate * estimate - 2.0)}
+
+        evaluate(improve())
+
+    def accurate_enough(*, loop):
+        return loop.result["error"] < 0.000001
+
+    @task
+    def finished():
+        print("Refinement finished.")
+
+    refinement = refine.loop(max_iterations=10, until=accurate_enough)
+    refinement >> finished()
+
+
+refine_estimate()
+# [END refine_estimate]
+
+
+# [START fixed_loop]
+@dag(schedule=None, catchup=False, tags=["example"])
+def fixed_task_loop():
+    @task_group
+    def accumulate():
+        @task
+        def increment(*, loop):
+            return (0 if loop.index == 0 else loop.previous) + 1
+
+        increment()
+
+    accumulate.loop(max_iterations=3)
+
+
+fixed_task_loop()
+# [END fixed_loop]
+
+
+# [START mapped_loop]
+@dag(schedule=None, catchup=False, tags=["example"])
+def mapped_task_loop():
+    @task_group
+    def process_batch():
+        @task
+        def choose_items(*, loop):
+            return [1, 2] if loop.index == 0 else [value + 1 for value in loop.previous]
+
+        @task
+        def process(value, *, loop, ti):
+            print(f"Iteration {loop.index}, mapped position {ti.map_index}")
+            return value + loop.index
+
+        process.expand(value=choose_items())
+
+    def batch_ready(*, loop):
+        return min(loop.result) >= 2
+
+    process_batch.loop(max_iterations=3, until=batch_ready)
+
+
+mapped_task_loop()
+# [END mapped_loop]
+
+
+# [START partial_override_loop]
+@dag(schedule=None, catchup=False, tags=["example"])
+def partial_override_loop():
+    @task_group
+    def accumulate(increment_by):
+        @task
+        def increment(increment_by, *, loop):
+            return (0 if loop.index == 0 else loop.previous) + increment_by
+
+        increment(increment_by)
+
+    accumulate.override(group_id="accumulate_more").partial(increment_by=5).loop(max_iterations=3)
+
+
+partial_override_loop()
+# [END partial_override_loop]

airflow-core/docs/authoring-and-scheduling/index.rst

@@ -31,7 +31,9 @@ It's recommended that you first review the pages in :doc:`core concepts </core-c
     deferring
     serializers
     connections
+    loops-and-mapped-tasks
     dynamic-task-mapping
+    loops
     assets
     dag-result
     Non-Python Task SDKs <language-sdks/index>

airflow-core/docs/authoring-and-scheduling/loops-and-mapped-tasks.rst

@@ -0,0 +1,61 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+.. _loops-and-mapped-tasks:
+
+======================
+Loops and mapped tasks
+======================
+
+A Dag does not have to be static. How much work it does can depend on data that only exists at
+runtime: how many files arrived, what the previous step found, or whether an answer is good enough
+yet. Airflow has two features for this, as well as mechanisms like ``retries`` and dynamic Dag generation that fit adjacent use cases.
+
+Choose a mechanism
+==================
+
+.. list-table::
+   :header-rows: 1
+   :widths: 30 30 40
+
+   * - You want to
+     - Use
+     - How it works
+   * - Run the same task once for each item in a collection
+     - :ref:`Mapped tasks <mapped-tasks>`
+     - ``task.expand(...)`` creates one task instance per item once the collection is known. The
+       instances can run concurrently.
+   * - Repeat a group of tasks, each iteration building on the result of the previous one, until a
+       condition is met
+     - :ref:`Loops <loops>`
+     - ``task_group.loop(...)`` creates the next iteration of task instances only if the last iteration says
+       another is needed.
+   * - Give a failed task another try
+     - Retries, set with ``retries`` on the task. See :doc:`/core-concepts/tasks`.
+     - A retry reruns the same task instance. It does not create new task instances, and it does not advance a
+       loop.
+   * - Create tasks from something known when the Dag file is parsed
+     - A Python ``for`` loop in the Dag file. See :doc:`/howto/dynamic-dag-generation`.
+     - The Dag has the same shape in every run.
+
+Use them together
+=================
+
+An iteration of a loop can contain mapped tasks, so each iteration can fan out over a different collection of
+items. See :ref:`Mapped tasks inside a loop <loops-mapped-tasks>`.
+
+Mapping a whole loop, nesting loops, and placing a loop inside a mapped task group are not supported.