code: def exec(self) -> Iterator[Batch]:
      from ray.util.queue import Queue
  
      input_queue = Queue(maxsize=100)
      output_queue = Queue(maxsize=100)
  
      # Pull data from child executor
      assert (
          len(self.children) == 1
      ), "Exchange currently only supports parallelization of node with only one child"
      ray_pull_task = ray_pull().remote(
          self.ray_pull_env_conf_dict,
          self.children[0],
          input_queue,
      )
  
      # Parallel the inner executor.
      ray_parallel_task_list = []
      for i in range(self.parallelism):
          ray_parallel_task_list.append(
              ray_parallel().remote(
                  self.ray_parallel_env_conf_dict[i],
                  self.inner_executor,
                  input_queue,
                  output_queue,
              )
          )
  
      ray_wait_and_alert().remote([ray_pull_task], input_queue)
      ray_wait_and_alert().remote(ray_parallel_task_list, output_queue)
  
      while True:
          res = output_queue.get(block=True)
          if res is StageCompleteSignal:
              break
          elif isinstance(res, ExecutorError):
              raise res
          else:
              yield res
  
file_path: evadb/evadb/executor/exchange_executor.py
imports: from typing import Iterator
  from evadb.database import EvaDBDatabase
  from evadb.executor.abstract_executor import AbstractExecutor
  from evadb.executor.executor_utils import ExecutorError
  from evadb.executor.ray_utils import StageCompleteSignal, ray_parallel, ray_pull, ray_wait_and_alert
  from evadb.models.storage.batch import Batch
  from evadb.plan_nodes.exchange_plan import ExchangePlan
label_desc: Use image node_id 3 for calling the ExchangeExecutor obj's underlying member method code with example usage: obj.exec() without return types