"""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:
            
            
            
            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.query(TaskInstance)