ashb commented on code in PR #74346:
URL: https://github.com/apache/airflow/pull/74346#discussion_r4216429688
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -2092,38 +2135,50 @@ def create_ti(task: Operator, indexes: Iterable[int])
-> Iterator[TI]:
def _create_tasks(
self,
tasks: Iterable[Operator],
- task_creator: Callable[[Operator, Iterable[int]], CreatedTasks],
+ task_creator: Callable[[Operator, Iterable[int], UUID], CreatedTasks],
*,
session: Session,
+ parent_region: tuple[UUID, int] | None = None,
+ expand_literals: bool = False,
) -> CreatedTasks:
"""
- Create missing tasks -- and expand any MappedOperator that _only_ have
literals as input.
+ Create ordinary tasks and a region-owned placeholder for each mapped
task.
:param tasks: Tasks to create jobs for in the DAG run
:param task_creator: Function to create task instances
+ :param expand_literals: Create the slots of a task whose inputs are
all literals right away,
+ so its placeholder is never expanded one task at a time.
"""
- from airflow.models.expandinput import NotFullyPopulated
- from airflow.serialization.definitions.mappedoperator import
get_mapped_ti_count
-
- map_indexes: Iterable[int]
+ tasks = list(tasks)
+ regions = {
Review Comment:
Ooooh yes interesting one. Let me see if I can think if a db uniqueness
constraint that covers this...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
