code: def _iter_subdag_run_ids(
      dag: DAG,
      session: SASession,
      state: DagRunState,
      task_ids: list[str],
      commit: bool,
      confirmed_infos: Iterable[_DagRunInfo],
  ) -> Iterator[str]:
      """Go through subdag operators and create dag runs.
  
      We only work within the scope of the subdag. A subdag does not propagate to
      its parent DAG, but parent propagates to subdags.
      """
      dags = [dag]
      while dags:
          current_dag = dags.pop()
          for task_id in task_ids:
              if not current_dag.has_task(task_id):
                  continue
  
              current_task = current_dag.get_task(task_id)
              if (
                  isinstance(current_task, SubDagOperator)
                  or current_task.task_type == "SubDagOperator"
              ):
                  # this works as a kind of integrity check
                  # it creates missing dag runs for subdag operators,
                  # maybe this should be moved to dagrun.verify_integrity
                  if TYPE_CHECKING:
                      assert current_task.subdag
                  dag_runs = _create_dagruns(
                      current_task.subdag,
                      infos=confirmed_infos,
                      state=DagRunState.RUNNING,
                      run_type=DagRunType.BACKFILL_JOB,
                  )
  
                  verify_dagruns(
                      dag_runs, commit, state, session, current_task
                  )
  
                  dags.append(current_task.subdag)
                  yield current_task.subdag.dag_id
  
imports: from __future__ import annotations
  from typing import TYPE_CHECKING, Collection, Iterable, Iterator, NamedTuple
  from sqlalchemy import or_, select
  from sqlalchemy.orm import lazyload
  from airflow.models.dagrun import DagRun
  from airflow.models.taskinstance import TaskInstance
  from airflow.operators.subdag import SubDagOperator
  from airflow.utils import timezone
  from airflow.utils.helpers import exactly_one
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import DagRunState, State, TaskInstanceState
  from airflow.utils.types import DagRunType
label_desc: Use image node_id 5 for calling a global function with example usage: _iter_subdag_run_ids(dag, session, state, task_ids, commit, confirmed_infos) without return types