code: def test_exception_concise_stacktrace():
      driver_script = """
  import ray
  
  def map(_):
      raise ValueError("foo")
  
  ray.data.range(1).map(map).take_all()
      """
      proc = run_string_as_driver_nonblocking(driver_script)
      out_str = proc.stdout.read().decode(
          "utf-8"
      ) + proc.stderr.read().decode("utf-8")
      # Test that the stack trace only contains the UDF exception, but not any other
      # exceptions raised when the executor is handling the UDF exception.
      assert (
          "During handling of the above exception, another exception occurred"
          not in out_str
      ), out_str
  
imports: import collections
  import time
  from unittest.mock import MagicMock
  import pytest
  import ray
  from ray._private.test_utils import run_string_as_driver_nonblocking, wait_for_condition
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, PhysicalOperator
  from ray.data._internal.execution.interfaces.physical_operator import MetadataOpTask
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import create_map_transformer_from_block_fn
  from ray.data._internal.execution.streaming_executor import _debug_dump_topology, _validate_dag
  from ray.data._internal.execution.streaming_executor_state import AutoscalingState, DownstreamMemoryInfo, OpState, TopologyResourceUsage, _execution_allowed, build_streaming_topology, process_completed_tasks, select_operator_to_run, update_operator_states
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.tests.conftest import *
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 18 for calling a global function with example usage: test_exception_concise_stacktrace() without return types