class_docstr: Common implementation for operators, including unmapped and mapped.
  
  This base class is more about sharing implementations, not defining a common
  interface. Unfortunately it's difficult to use this as the common base class
  for typing due to BaseOperator carrying too much historical baggage.
  
  The union type ``from airflow.models.operator import Operator`` is easier
  to use for typing purposes.
  
  :meta private:
class_docstr_tok: ["Common","implementation","for","operators",",","including","unmapped","and","mapped",".","This","base","class","is","more","about","sharing","implementations",",","not","defining","a","common","interface",".","Unfortunately","it","'s","difficult","to","use","this","as","the","common","base","class","for","typing","due","to","BaseOperator","carrying","too","much","historical","baggage",".","The","union","type","``","from","airflow.models.operator","import","Operator","``","is","easier","to","use","for","typing","purposes",".",":","meta","private",":"]
docstr: Create the mapped task instances for mapped task.
  
  :raise NotMapped: If this task does not need expansion.
  :return: The newly created mapped task instances (if any) in ascending
      order by map index, and the maximum map index value.
docstr_tok: ["Create","the","mapped","task","instances","for","mapped","task",".",":","raise","NotMapped",":","If","this","task","does","not","need","expansion",".",":","return",":","The","newly","created","mapped","task","instances","(","if","any",")","in","ascending","order","by","map","index",",","and","the","maximum","map","index","value","."]
code: def expand_mapped_task(
      self, run_id: str, *, session: Session
  ) -> tuple[Sequence[TaskInstance], int]:
      """Create the mapped task instances for mapped task.
  
      :raise NotMapped: If this task does not need expansion.
      :return: The newly created mapped task instances (if any) in ascending
          order by map index, and the maximum map index value.
      """
      from sqlalchemy import func, or_
  
      from airflow.models.baseoperator import BaseOperator
      from airflow.models.mappedoperator import MappedOperator
      from airflow.models.taskinstance import TaskInstance
      from airflow.settings import task_instance_mutation_hook
  
      if not isinstance(self, (BaseOperator, MappedOperator)):
          raise RuntimeError(
              f"cannot expand unrecognized operator type {type(self).__name__}"
          )
  
      try:
          total_length: int | None = self.get_mapped_ti_count(
              run_id, session=session
          )
      except NotFullyPopulated as e:
          # It's possible that the upstream tasks are not yet done, but we
          # don't have upstream of upstreams in partial DAGs (possible in the
          # mini-scheduler), so we ignore this exception.
          if not self.dag or not self.dag.partial:
              self.log.error(
                  "Cannot expand %r for run %s; missing upstream values: %s",
                  self,
                  run_id,
                  sorted(e.missing),
              )
          total_length = None
  
      state: TaskInstanceState | None = None
      unmapped_ti: TaskInstance | None = session.scalars(
          select(TaskInstance).where(
              TaskInstance.dag_id == self.dag_id,
              TaskInstance.task_id == self.task_id,
              TaskInstance.run_id == run_id,
              TaskInstance.map_index == -1,
              or_(
                  TaskInstance.state.in_(State.unfinished),
                  TaskInstance.state.is_(None),
              ),
          )
      ).one_or_none()
  
      all_expanded_tis: list[TaskInstance] = []
  
      if unmapped_ti:
          # The unmapped task instance still exists and is unfinished, i.e. we
          # haven't tried to run it before.
          if total_length is None:
              # If the DAG is partial, it's likely that the upstream tasks
              # are not done yet, so the task can't fail yet.
              if not self.dag or not self.dag.partial:
                  unmapped_ti.state = TaskInstanceState.UPSTREAM_FAILED
          elif total_length < 1:
              # If the upstream maps this to a zero-length value, simply mark
              # the unmapped task instance as SKIPPED (if needed).
              self.log.info(
                  "Marking %s as SKIPPED since the map has %d values to expand",
                  unmapped_ti,
                  total_length,
              )
              unmapped_ti.state = TaskInstanceState.SKIPPED
          else:
              zero_index_ti_exists = exists_query(
                  TaskInstance.dag_id == self.dag_id,
                  TaskInstance.task_id == self.task_id,
                  TaskInstance.run_id == run_id,
                  TaskInstance.map_index == 0,
                  session=session,
              )
              if not zero_index_ti_exists:
                  # Otherwise convert this into the first mapped index, and create
                  # TaskInstance for other indexes.
                  unmapped_ti.map_index = 0
                  self.log.debug(
                      "Updated in place to become %s", unmapped_ti
                  )
                  all_expanded_tis.append(unmapped_ti)
                  session.flush()
              else:
                  self.log.debug(
                      "Deleting the original task instance: %s",
                      unmapped_ti,
                  )
                  session.delete(unmapped_ti)
              state = unmapped_ti.state
  
      if total_length is None or total_length < 1:
          # Nothing to fixup.
          indexes_to_map: Iterable[int] = ()
      else:
          # Only create "missing" ones.
          current_max_mapping = session.scalar(
              select(func.max(TaskInstance.map_index)).where(
                  TaskInstance.dag_id == self.dag_id,
                  TaskInstance.task_id == self.task_id,
                  TaskInstance.run_id == run_id,
              )
          )
          indexes_to_map = range(current_max_mapping + 1, total_length)
  
      for index in indexes_to_map:
          # TODO: Make more efficient with bulk_insert_mappings/bulk_save_mappings.
          ti = TaskInstance(
              self, run_id=run_id, map_index=index, state=state
          )
          self.log.debug("Expanding TIs upserted %s", ti)
          task_instance_mutation_hook(ti)
          ti = session.merge(ti)
          ti.refresh_from_task(
              self
          )  # session.merge() loses task information.
          all_expanded_tis.append(ti)
  
      # Coerce the None case to 0 -- these two are almost treated identically,
      # except the unmapped ti (if exists) is marked to different states.
      total_expanded_ti_count = total_length or 0
  
      # Any (old) task instances with inapplicable indexes (>= the total
      # number we need) are set to "REMOVED".
      query = select(TaskInstance).where(
          TaskInstance.dag_id == self.dag_id,
          TaskInstance.task_id == self.task_id,
          TaskInstance.run_id == run_id,
          TaskInstance.map_index >= total_expanded_ti_count,
      )
      query = with_row_locks(
          query,
          of=TaskInstance,
          session=session,
          **skip_locked(session=session),
      )
      to_update = session.scalars(query)
      for ti in to_update:
          ti.state = TaskInstanceState.REMOVED
      session.flush()
      return all_expanded_tis, total_expanded_ti_count - 1
  
imports: from __future__ import annotations
  import datetime
  import inspect
  from functools import cached_property
  from typing import TYPE_CHECKING, Any, Callable, ClassVar, Collection, Iterable, Iterator, Sequence
  from sqlalchemy import select
  from airflow.compat.functools import cache
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException
  from airflow.models.expandinput import NotFullyPopulated
  from airflow.models.taskmixin import DAGNode, DependencyMixin
  from airflow.template.templater import Templater
  from airflow.utils.context import Context
  from airflow.utils.db import exists_query
  from airflow.utils.log.secrets_masker import redact
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.setup_teardown import SetupTeardownContext
  from airflow.utils.sqlalchemy import skip_locked, with_row_locks
  from airflow.utils.state import State, TaskInstanceState
  from airflow.utils.task_group import MappedTaskGroup
  from airflow.utils.trigger_rule import TriggerRule
  from airflow.utils.types import NOTSET, ArgNotSet
  from airflow.utils.weight_rule import WeightRule
label_desc: Use image node_id 33 for calling the AbstractOperator obj's underlying member method code with example usage: obj.expand_mapped_task(run_id) and returns: all_expanded_tis,  unknown