code: def run_op_tasks_sync(op: PhysicalOperator, only_existing=False):
      """Run tasks of a PhysicalOperator synchronously.
  
      By default, this function will run until the op no longer has any active tasks.
      If only_existing is True, this function will only run the currently existing tasks.
      """
      tasks = op.get_active_tasks()
      while tasks:
          ray.wait(
              [task.get_waitable() for task in tasks],
              num_returns=len(tasks),
              fetch_local=False,
          )
          for task in tasks:
              if isinstance(task, DataOpTask):
                  task.on_data_ready(None)
              else:
                  assert isinstance(task, MetadataOpTask)
                  task.on_task_finished()
          if only_existing:
              return
          tasks = op.get_active_tasks()
  
file_path: ray/python/ray/data/tests/util.py
imports: import functools
  import os
  import tempfile
  from contextlib import contextmanager
  import ray
  from ray.data._internal.execution.interfaces.physical_operator import DataOpTask, MetadataOpTask, PhysicalOperator
label_desc: Use image node_id 6 for calling a global function with example usage: run_op_tasks_sync(op, only_existing) without return types