code: def select_operator_to_run(
      topology: Topology,
      cur_usage: TopologyResourceUsage,
      limits: ExecutionResources,
      backpressure_policies: List[BackpressurePolicy],
      ensure_at_least_one_running: bool,
      execution_id: str,
      autoscaling_state: AutoscalingState,
  ) -> Optional[PhysicalOperator]:
      """Select an operator to run, if possible.
  
      The objective of this function is to maximize the throughput of the overall
      pipeline, subject to defined memory and parallelism limits.
  
      This is currently implemented by applying backpressure on operators that are
      producing outputs faster than they are consuming them `len(outqueue)`, as well as
      operators with a large number of running tasks `num_processing()`.
  
      Note that memory limits also apply to the outqueue of the output operator. This
      provides backpressure if the consumer is slow. However, once a bundle is returned
      to the user, it is no longer tracked.
      """
      assert isinstance(cur_usage, TopologyResourceUsage), cur_usage
  
      # Filter to ops that are eligible for execution.
      ops = []
      for op, state in topology.items():
          under_resource_limits = _execution_allowed(
              op, cur_usage, limits
          )
          if (
              op.need_more_inputs()
              and state.num_queued() > 0
              and op.should_add_input()
              and under_resource_limits
              and not op.completed()
              and all(
                  p.can_add_input(op) for p in backpressure_policies
              )
          ):
              ops.append(op)
          # Update the op in all cases to enable internal autoscaling, etc.
          op.notify_resource_usage(
              state.num_queued(), under_resource_limits
          )
  
      # If no ops are allowed to execute due to resource constraints, try to trigger
      # cluster scale-up.
      if not ops and any(
          state.num_queued() > 0 for state in topology.values()
      ):
          now = time.time()
          if (
              now
              > autoscaling_state.last_request_ts
              + MIN_GAP_BETWEEN_AUTOSCALING_REQUESTS
          ):
              autoscaling_state.last_request_ts = now
              _try_to_scale_up_cluster(topology, execution_id)
  
      # To ensure liveness, allow at least 1 op to run regardless of limits. This is
      # gated on `ensure_at_least_one_running`, which is set if the consumer is blocked.
      if (
          ensure_at_least_one_running
          and not ops
          and all(op.num_active_tasks() == 0 for op in topology)
      ):
          # The topology is entirely idle, so choose from all ready ops ignoring limits.
          ops = [
              op
              for op, state in topology.items()
              if op.need_more_inputs()
              and state.num_queued() > 0
              and not op.completed()
          ]
  
      # Nothing to run.
      if not ops:
          return None
  
      # Run metadata-only operators first. After that, equally penalize outqueue length
      # and num bundles processing for backpressure.
      return min(
          ops,
          key=lambda op: (
              not op.throttling_disabled(),
              len(topology[op].outqueue)
              + topology[op].num_processing(),
          ),
      )
  
imports: import math
  import threading
  import time
  from collections import defaultdict, deque
  from dataclasses import dataclass
  from typing import Dict, List, Optional, Tuple
  import ray
  from ray.data._internal.dataset_logger import DatasetLogger
  from ray.data._internal.execution.autoscaling_requester import get_or_create_autoscaling_requester_actor
  from ray.data._internal.execution.backpressure_policy import BackpressurePolicy
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, PhysicalOperator, RefBundle
  from ray.data._internal.execution.interfaces.physical_operator import DataOpTask, MetadataOpTask, OpTask, Waitable
  from ray.data._internal.execution.operators.base_physical_operator import AllToAllOperator
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.util import memory_string
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data.context import DataContext
label_desc: Use image node_id 4 for calling a global function with example usage: select_operator_to_run(topology, cur_usage, limits, backpressure_policies, ensure_at_least_one_running, execution_id, autoscaling_state) and returns: min, None