code: def test_parquet(s3, engine, s3so, metadata_file):
      dd = pytest.importorskip("dask.dataframe")
      pd = pytest.importorskip("pandas")
      np = pytest.importorskip("numpy")
  
      url = "s3://%s/test.parquet" % test_bucket_name
      data = pd.DataFrame(
          {
              "i32": np.arange(1000, dtype=np.int32),
              "i64": np.arange(1000, dtype=np.int64),
              "f": np.arange(1000, dtype=np.float64),
              "bhello": np.random.choice(
                  ["hello", "you", "people"], size=1000
              ).astype("O"),
          },
          index=pd.Index(np.arange(1000), name="foo"),
      )
      df = dd.from_pandas(data, chunksize=500)
      df.to_parquet(
          url,
          engine=engine,
          storage_options=s3so,
          write_metadata_file=metadata_file,
      )
  
      files = [f.split("/")[-1] for f in s3.ls(url)]
      if metadata_file:
          assert "_common_metadata" in files
          assert "_metadata" in files
      assert "part.0.parquet" in files
  
      df2 = dd.read_parquet(
          url,
          index="foo",
          calculate_divisions=True,
          engine=engine,
          storage_options=s3so,
      )
      assert len(df2.divisions) > 1
  
      dd.utils.assert_eq(data, df2)
  
      # Check that `open_file_options` arguments are
      # really passed through to fsspec
      if fsspec_parquet:
          # Passing `open_file_options` kwargs will fail
          # if you set an unsupported engine
          with pytest.raises(ValueError):
              dd.read_parquet(
                  url,
                  engine=engine,
                  storage_options=s3so,
                  open_file_options={
                      "precache_options": {
                          "method": "parquet",
                          "engine": "foo",
                      },
                  },
              ).compute()
  
          # ...but should work fine if you modify the
          # maximum block-transfer size (max_block)
          dd.read_parquet(
              url,
              engine=engine,
              storage_options=s3so,
              open_file_options={
                  "precache_options": {
                      "method": "parquet",
                      "max_block": 8_000,
                  },
              },
          ).compute()
  
      # Check "open_file_func"
      fs = get_fs_token_paths(url, storage_options=s3so)[0]
  
      def _open(*args, check=True, **kwargs):
          assert check
          return fs.open(*args, **kwargs)
  
      # Should fail if `check=False`
      with pytest.raises(AssertionError):
          dd.read_parquet(
              url,
              engine=engine,
              storage_options=s3so,
              open_file_options={
                  "open_file_func": _open,
                  "check": False,
              },
          ).compute()
  
      # Should succeed otherwise
      df3 = dd.read_parquet(
          url,
          engine=engine,
          storage_options=s3so,
          open_file_options={"open_file_func": _open},
      )
      dd.utils.assert_eq(data, df3)
  
      # Check that `cache_type="all"` result is same
      df4 = dd.read_parquet(
          url,
          engine=engine,
          storage_options=s3so,
          open_file_options={"cache_type": "all"},
      )
      dd.utils.assert_eq(data, df4)
  
imports: from __future__ import annotations
  import io
  import os
  import shlex
  import subprocess
  import sys
  import time
  from contextlib import contextmanager
  from functools import partial
  import pytest
  from fsspec.compression import compr
  from fsspec.core import get_fs_token_paths, open_files
  from s3fs import S3FileSystem
  from tlz import concat, valmap
  from dask import compute
  from dask.bytes.core import read_bytes
  from dask.bytes.utils import compress
label_desc: Use image node_id 20 for calling a global function with example usage: test_parquet(s3, engine, s3so, metadata_file) and returns: fs