code: def test_map_with_mismatched_columns(ray_start_regular_shared):
      def bad_fn(row):
          if row["id"] > 5:
              return {"a": "hello1"}
          else:
              return {"b": "hello1"}
  
      def good_fn(row):
          if row["id"] > 5:
              return {"a": "hello1", "b": "hello2"}
          else:
              return {"b": "hello2", "a": "hello1"}
  
      ds = ray.data.range(10, parallelism=1)
      error_message = (
          "Current row has different columns compared to previous rows."
      )
      with pytest.raises(ValueError) as e:
          ds.map(bad_fn).materialize()
      assert error_message in str(e.value)
      ds_map = ds.map(good_fn)
      assert ds_map.take() == [
          {"a": "hello1", "b": "hello2"} for _ in range(10)
      ]
  
imports: import itertools
  import math
  import os
  import signal
  import threading
  import time
  from typing import Iterator
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import column_udf, column_udf_class, extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 19 for calling a global function with example usage: test_map_with_mismatched_columns(ray_start_regular_shared) and returns: dict, dict, dict, dict