code: def test_csv_read_many_files_partitioned(
      ray_start_regular_shared,
      fs,
      data_path,
      endpoint_url,
      write_partitioned_df,
      assert_base_partitioned_ds,
  ):
      if endpoint_url is None:
          storage_options = {}
      else:
          storage_options = dict(
              client_kwargs=dict(endpoint_url=endpoint_url)
          )
  
      partition_keys = ["one"]
      partition_path_encoder = PathPartitionEncoder.of(
          base_dir=data_path,
          field_names=partition_keys,
          filesystem=fs,
      )
      paths = []
      dfs = []
      num_dfs = FILE_SIZE_FETCH_PARALLELIZATION_THRESHOLD
      num_rows = 6 * num_dfs
      num_files = 2 * num_dfs
      for i in range(num_dfs):
          df = pd.DataFrame(
              {
                  "one": [1, 1, 1, 3, 3, 3],
                  "two": list(range(6 * i, 6 * (i + 1))),
              }
          )
          df_paths = write_partitioned_df(
              df,
              partition_keys,
              partition_path_encoder,
              partial(
                  df_to_csv,
                  storage_options=storage_options,
                  index=False,
              ),
              file_name_suffix=i,
          )
          dfs.append(df)
          paths.extend(df_paths)
  
      ds = ray.data.read_csv(
          paths,
          filesystem=fs,
          partitioning=partition_path_encoder.scheme,
          parallelism=num_files,
      )
  
      assert_base_partitioned_ds(
          ds,
          count=num_rows,
          num_input_files=num_files,
          num_rows=num_rows,
          schema="{one: int64, two: int64}",
          num_computed=num_files,
          sorted_values=sorted(
              itertools.chain.from_iterable(
                  list(
                      map(
                          list,
                          zip(
                              [1, 1, 1, 3, 3, 3],
                              list(range(6 * i, 6 * (i + 1))),
                          ),
                      )
                  )
                  for i in range(num_dfs)
              )
          ),
      )
  
imports: import itertools
  import os
  import shutil
  from functools import partial
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  from packaging.version import Version
  from pytest_lazyfixture import lazy_fixture
  import ray
  from ray.data.block import BlockAccessor
  from ray.data.datasource import BaseFileMetadataProvider, FastFileMetadataProvider, PartitionStyle, PathPartitionFilter
  from ray.data.datasource.file_based_datasource import FILE_SIZE_FETCH_PARALLELIZATION_THRESHOLD
  from ray.data.datasource.path_util import _unwrap_protocol
  from ray.data.tests.conftest import *
  from ray.data.tests.mock_http_server import *
  from ray.data.tests.test_partitioning import PathPartitionEncoder
  from ray.data.tests.util import Counter
  from ray.tests.conftest import *
label_desc: Use image node_id 7 for calling a global function with example usage: test_csv_read_many_files_partitioned(ray_start_regular_shared, fs, data_path, endpoint_url, write_partitioned_df, assert_base_partitioned_ds) without return types