code: def test_fsspec_filesystem(ray_start_regular_shared, tmp_path):
      """Same as `test_parquet_write` but using a custom, fsspec filesystem.
  
      TODO (Alex): We should write a similar test with a mock PyArrow fs, but
      unfortunately pa.fs._MockFileSystem isn't serializable, so this may require
      some effort.
      """
      df1 = pd.DataFrame({"one": [1, 2, 3], "two": ["a", "b", "c"]})
      table = pa.Table.from_pandas(df1)
      path1 = os.path.join(str(tmp_path), "test1.parquet")
      pq.write_table(table, path1)
      df2 = pd.DataFrame({"one": [4, 5, 6], "two": ["e", "f", "g"]})
      table = pa.Table.from_pandas(df2)
      path2 = os.path.join(str(tmp_path), "test2.parquet")
      pq.write_table(table, path2)
  
      fs = LocalFileSystem()
  
      ds = ray.data.read_parquet([path1, path2], filesystem=fs)
  
      # Test metadata-only parquet ops.
      assert ds._plan.execute()._num_computed() == 0
      assert ds.count() == 6
  
      out_path = os.path.join(tmp_path, "out")
      os.mkdir(out_path)
  
      ds._set_uuid("data")
      ds.write_parquet(out_path)
  
      ds_df1 = pd.read_parquet(
          os.path.join(out_path, "data_000000_000000.parquet")
      )
      ds_df2 = pd.read_parquet(
          os.path.join(out_path, "data_000001_000000.parquet")
      )
      ds_df = pd.concat([ds_df1, ds_df2])
      df = pd.concat([df1, df2])
      assert ds_df.equals(df)
  
imports: import os
  from typing import Iterable, List
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  import torchvision
  from fsspec.implementations.http import HTTPFileSystem
  from fsspec.implementations.local import LocalFileSystem
  import ray
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.execution.interfaces import TaskContext
  from ray.data.block import Block, BlockAccessor
  from ray.data.datasource import Datasource, DummyOutputDatasource, WriteResult
  from ray.data.datasource.file_meta_provider import _handle_read_os_error
  from ray.data.tests.conftest import *
  from ray.data.tests.mock_http_server import *
  from ray.data.tests.util import extract_values
  from ray.tests.conftest import *
  from ray.types import ObjectRef
label_desc: Use image node_id 6 for calling a global function with example usage: test_fsspec_filesystem(ray_start_regular_shared, tmp_path) without return types