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 total number of output bundles of this operator.
  
  The value returned may be an estimate based off the consumption so far.
  This is useful for reporting progress.
docstr_tok: ["Returns","the","total","number","of","output","bundles","of","this","operator",".","The","value","returned","may","be","an","estimate","based","off","the","consumption","so","far",".","This","is","useful","for","reporting","progress","."]
code: def num_outputs_total(self) -> int:
      """Returns the total number of output bundles of this operator.
  
      The value returned may be an estimate based off the consumption so far.
      This is useful for reporting progress.
      """
      if self._estimated_output_blocks is not None:
          return self._estimated_output_blocks
      if len(self.input_dependencies) == 1:
          return self.input_dependencies[0].num_outputs_total()
      raise AttributeError
  
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 11 for calling the PhysicalOperator obj's underlying member method code with example usage: obj.num_outputs_total() and returns: self, self