code: def test_roundtrip_hive_parquet_spark_to_dask(
      spark_session, tmpdir, engine
  ):
      tmpdir = str(tmpdir)
  
      sdf = spark_session.createDataFrame(pdf)
      # not overwriting any data, but spark complains if the directory
      # already exists and we don't set overwrite
      sdf.write.parquet(tmpdir, mode="overwrite", partitionBy="name")
  
      ddf = dd.read_parquet(tmpdir, engine=engine)
      # Papercut: pandas TZ localization doesn't survive roundtrip
      ddf = ddf.assign(timestamp=ddf.timestamp.dt.tz_localize("UTC"))
  
      # Partitioning can change the column order. This is mostly okay,
      # but we sort them here to ease comparison
      ddf = ddf.compute().sort_index(axis=1)
      # Dask automatically converts hive-partitioned columns to categories.
      # This is fine, but convert back to strings for comparison.
      ddf = ddf.assign(name=ddf.name.astype("str"))
  
      assert_eq(ddf, pdf.sort_index(axis=1), check_index=False)
  
file_path: dask/dask/tests/test_spark_compat.py
imports: from __future__ import annotations
  import decimal
  import signal
  import sys
  import threading
  import pytest
  from dask.datasets import timeseries
  import numpy
  import pandas
  from dask.dataframe._compat import PANDAS_GE_150, PANDAS_GE_200
  from dask.dataframe.utils import assert_eq
label_desc: Use image node_id 4 for calling a global function with example usage: test_roundtrip_hive_parquet_spark_to_dask(spark_session, tmpdir, engine) without return types