code: def execute(self):
          
          self._collect_garbage()  # Make sure everything is clean
          # TODO (john): Check if dataflow has any 'logical inconsistencies'
          # For example, if there is a forward partitioning strategy but
          # the number of downstream instances is larger than the number of
          # upstream instances, some of the downstream instances will not be
          # used at all
  
          # Each operator instance is implemented as a Ray actor
          # Actors are deployed in topological order, as we traverse the
          # logical dataflow from sources to sinks. At each step, data
          # producers wait for acknowledge from consumers before starting
          # generating data.
          upstream_channels = {}
          for node in nx.topological_sort(self.logical_topo):
              operator = self.operators[node]
              # Generate downstream data channels
              downstream_channels = self._generate_channels(operator)
              # Instantiate Ray actors
              handles = self.__generate_actors(operator, upstream_channels,
                                               downstream_channels)
              if handles:
                  self.actor_handles.extend(handles)
              upstream_channels.update(downstream_channels)
          logger.debug("Running...")
          return self.actor_handles
docs: Deploys and executes the physical dataflow.