class_docstr: Abstract class for physical operators.
  
  An operator transforms one or more input streams of RefBundles into a single
  output stream of RefBundles.
  
  Physical operators are stateful and non-serializable; they live on the driver side
  of the Dataset only.
  
  Here's a simple example of implementing a basic "Map" operator:
  
      class MapOperator(PhysicalOperator):
          def __init__(self):
              self.active_tasks = []
  
          def add_input(self, refs, _):
              self.active_tasks.append(map_task.remote(refs))
  
          def has_next(self):
              ready, _ = ray.wait(self.active_tasks, timeout=0)
              return len(ready) > 0
  
          def get_next(self):
              ready, remaining = ray.wait(self.active_tasks, num_returns=1)
              self.active_tasks = remaining
              return ready[0]
  
  Note that the above operator fully supports both bulk and streaming execution,
  since `add_input` and `get_next` can be called in any order. In bulk execution,
  all inputs would be added up-front, but in streaming execution the calls could
  be interleaved.
class_docstr_tok: ["Abstract","class","for","physical","operators",".","An","operator","transforms","one","or","more","input","streams","of","RefBundles","into","a","single","output","stream","of","RefBundles",".","Physical","operators","are","stateful","and","non-serializable",";","they","live","on","the","driver","side","of","the","Dataset","only",".","Here","'s","a","simple","example","of","implementing","a","basic","``","Map","''","operator",":","class","MapOperator","(","PhysicalOperator",")",":","def","__init__","(","self",")",":","self.active_tasks","=","[","]","def","add_input","(","self",",","refs",",","_",")",":","self.active_tasks.append","(","map_task.remote","(","refs",")",")","def","has_next","(","self",")",":","ready",",","_","=","ray.wait","(","self.active_tasks",",","timeout=0",")","return","len","(","ready",")",">","0","def","get_next","(","self",")",":","ready",",","remaining","=","ray.wait","(","self.active_tasks",",","num_returns=1",")","self.active_tasks","=","remaining","return","ready","[","0","]","Note","that","the","above","operator","fully","supports","both","bulk","and","streaming","execution",",","since","`","add_input","`","and","`","get_next","`","can","be","called","in","any","order",".","In","bulk","execution",",","all","inputs","would","be","added","up-front",",","but","in","streaming","execution","the","calls","could","be","interleaved","."]
docstr: Returns the current estimated resource usage of this operator.
  
  This method is called by the executor to decide how to allocate resources
  between different operators.
docstr_tok: ["Returns","the","current","estimated","resource","usage","of","this","operator",".","This","method","is","called","by","the","executor","to","decide","how","to","allocate","resources","between","different","operators","."]
code: def current_resource_usage(self) -> ExecutionResources:
      """Returns the current estimated resource usage of this operator.
  
      This method is called by the executor to decide how to allocate resources
      between different operators.
      """
      return ExecutionResources()
  
imports: from abc import ABC, abstractmethod
  from typing import Any, Callable, Dict, List, Optional, Union
  import ray
  from .ref_bundle import RefBundle
  from ray._raylet import ObjectRefGenerator
  from ray.data._internal.execution.interfaces.execution_options import ExecutionOptions, ExecutionResources
  from ray.data._internal.execution.interfaces.op_runtime_metrics import OpRuntimeMetrics
  from ray.data._internal.logical.interfaces import Operator
  from ray.data._internal.stats import StatsDict
  from ray.data.context import DataContext
label_desc: Use image node_id 28 for calling the PhysicalOperator obj's underlying member method code with example usage: obj.current_resource_usage() and returns: ExecutionResources