class_docstr: A backfill job runner consists of a dag or subdag for a specific time range.
  
  It triggers a set of task instance runs, in the right order and lasts for
  as long as it takes for the set of task instance to be completed.
docstr: Reset state of orphaned tasks.
  
  This function checks if there are any tasks in the dagrun (or all) that
  have a schedule or queued states but are not known by the executor. If
  it finds those it will reset the state to None so they will get picked
  up again.  The batch option is for performance reasons as the queries
  are made in sequence.
  
  :param filter_by_dag_run: the dag_run we want to process, None if all
  :return: the number of TIs reset
docstr_tok: ["Reset","state","of","orphaned","tasks",".","This","function","checks","if","there","are","any","tasks","in","the","dagrun","(","or","all",")","that","have","a","schedule","or","queued","states","but","are","not","known","by","the","executor",".","If","it","finds","those","it","will","reset","the","state","to","None","so","they","will","get","picked","up","again",".","The","batch","option","is","for","performance","reasons","as","the","queries","are","made","in","sequence",".",":","param","filter_by_dag_run",":","the","dag_run","we","want","to","process",",","None","if","all",":","return",":","the","number","of","TIs","reset"]
code: def reset_state_for_orphaned_tasks(
      self,
      filter_by_dag_run: DagRun | None = None,
      session: Session = NEW_SESSION,
  ) -> int | None:
      """
      Reset state of orphaned tasks.
  
      This function checks if there are any tasks in the dagrun (or all) that
      have a schedule or queued states but are not known by the executor. If
      it finds those it will reset the state to None so they will get picked
      up again.  The batch option is for performance reasons as the queries
      are made in sequence.
  
      :param filter_by_dag_run: the dag_run we want to process, None if all
      :return: the number of TIs reset
      """
      queued_tis = self.job.executor.queued_tasks
      # also consider running as the state might not have changed in the db yet
      running_tis = self.job.executor.running
  
      # Can't use an update here since it doesn't support joins.
      resettable_states = [
          TaskInstanceState.SCHEDULED,
          TaskInstanceState.QUEUED,
      ]
      if filter_by_dag_run is None:
          resettable_tis = (
              session.scalars(
                  select(TaskInstance)
                  .join(TaskInstance.dag_run)
                  .where(
                      DagRun.state == DagRunState.RUNNING,
                      DagRun.run_type != DagRunType.BACKFILL_JOB,
                      TaskInstance.state.in_(resettable_states),
                  )
              )
          ).all()
      else:
          resettable_tis = filter_by_dag_run.get_task_instances(
              state=resettable_states, session=session
          )
  
      tis_to_reset = [
          ti
          for ti in resettable_tis
          if ti.key not in queued_tis and ti.key not in running_tis
      ]
      if not tis_to_reset:
          return 0
  
      def query(result, items):
          if not items:
              return result
  
          filter_for_tis = TaskInstance.filter_for_tis(items)
          reset_tis = session.scalars(
              select(TaskInstance)
              .where(
                  filter_for_tis,
                  TaskInstance.state.in_(resettable_states),
              )
              .with_for_update()
          ).all()
  
          for ti in reset_tis:
              ti.state = None
              session.merge(ti)
  
          return result + reset_tis
  
      reset_tis = helpers.reduce_in_chunks(
          query, tis_to_reset, [], self.job.max_tis_per_query
      )
  
      task_instance_str = "\n".join(f"\t{x!r}" for x in reset_tis)
      session.flush()
  
      self.log.info(
          "Reset the following %s TaskInstances:\n%s",
          len(reset_tis),
          task_instance_str,
      )
      return len(reset_tis)
  
imports: from __future__ import annotations
  import time
  from typing import TYPE_CHECKING, Any, Iterable, Iterator, Mapping, Sequence
  import attr
  import pendulum
  from sqlalchemy import select, tuple_, update
  from sqlalchemy.exc import OperationalError
  from sqlalchemy.orm.session import make_transient
  from tabulate import tabulate
  from airflow import models
  from airflow.exceptions import AirflowException, BackfillUnfinished, DagConcurrencyLimitReached, NoAvailablePoolSlot, PoolNotFound, TaskConcurrencyLimitReached
  from airflow.executors.executor_loader import ExecutorLoader
  from airflow.jobs.base_job_runner import BaseJobRunner
  from airflow.jobs.job import Job, perform_heartbeat
  from airflow.models import DAG, DagPickle
  from airflow.models.dagrun import DagRun
  from airflow.models.taskinstance import TaskInstance
  from airflow.ti_deps.dep_context import DepContext
  from airflow.ti_deps.dependencies_deps import BACKFILL_QUEUED_DEPS
  from airflow.timetables.base import DagRunInfo
  from airflow.utils import helpers, timezone
  from airflow.utils.configuration import conf, tmp_configuration_copy
  from airflow.utils.log.logging_mixin import LoggingMixin
  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 13 for calling the BackfillJobRunner obj's underlying member method code with example usage: obj.reset_state_for_orphaned_tasks(filter_by_dag_run, session) and returns: len, int, unknown, result