code: def test_default_file_metadata_provider_many_files_diff_dirs(
      ray_start_regular,
      propagate_logs,
      caplog,
      fs,
      data_path,
      endpoint_url,
  ):
      if endpoint_url is None:
          storage_options = {}
      else:
          storage_options = dict(
              client_kwargs=dict(endpoint_url=endpoint_url)
          )
  
      dir1 = os.path.join(data_path, "dir1")
      dir2 = os.path.join(data_path, "dir2")
      if fs is None:
          os.mkdir(dir1)
          os.mkdir(dir2)
      else:
          fs.create_dir(_unwrap_protocol(dir1))
          fs.create_dir(_unwrap_protocol(dir2))
  
      paths = []
      dfs = []
      num_dfs = 2 * FILE_SIZE_FETCH_PARALLELIZATION_THRESHOLD
      for i, dir_path in enumerate([dir1, dir2]):
          for j in range(num_dfs * i, num_dfs * (i + 1)):
              df = pd.DataFrame(
                  {"one": list(range(3 * j, 3 * (j + 1)))}
              )
              dfs.append(df)
              path = os.path.join(dir_path, f"test_{j}.csv")
              paths.append(path)
              df.to_csv(
                  path, index=False, storage_options=storage_options
              )
      paths, fs = _resolve_paths_and_filesystem(paths, fs)
  
      meta_provider = DefaultFileMetadataProvider()
      if isinstance(fs, LocalFileSystem):
          patcher = patch(
              "ray.data.datasource.file_meta_provider._get_file_infos_serial",
              wraps=_get_file_infos_serial,
          )
      else:
          patcher = patch(
              "ray.data.datasource.file_meta_provider._get_file_infos_parallel",
              wraps=_get_file_infos_parallel,
          )
      with caplog.at_level(logging.WARNING), patcher as mock_get:
          file_paths, file_sizes = map(
              list, zip(*meta_provider.expand_paths(paths, fs))
          )
  
      mock_get.assert_called_once_with(paths, fs, False)
      if isinstance(fs, LocalFileSystem):
          # No warning should be logged.
          assert len(caplog.text) == 0
      else:
          # Many files with different directories on cloud storage should log warning.
          assert "common parent directory" in caplog.text
      assert file_paths == paths
      expected_file_sizes = _get_file_sizes_bytes(paths, fs)
      assert file_sizes == expected_file_sizes
  
      # Many directories should not trigger error.
      if isinstance(fs, LocalFileSystem):
          dir_paths = [dir1, dir2] * num_dfs
          with caplog.at_level(logging.WARNING), patcher as mock_get:
              file_paths, file_sizes = map(
                  list, zip(*meta_provider.expand_paths(dir_paths, fs))
              )
          assert len(file_paths) == len(paths) * num_dfs
  
imports: import logging
  import os
  import posixpath
  import urllib.parse
  from functools import partial
  from unittest.mock import patch
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  from pyarrow.fs import LocalFileSystem
  from pytest_lazyfixture import lazy_fixture
  from ray.data.datasource import BaseFileMetadataProvider, DefaultFileMetadataProvider, DefaultParquetMetadataProvider, FastFileMetadataProvider, FileMetadataProvider, ParquetMetadataProvider
  from ray.data.datasource.file_based_datasource import FILE_SIZE_FETCH_PARALLELIZATION_THRESHOLD
  from ray.data.datasource.file_meta_provider import _get_file_infos_common_path_prefix, _get_file_infos_parallel, _get_file_infos_serial
  from ray.data.datasource.path_util import _resolve_paths_and_filesystem, _unwrap_protocol
  from ray.data.tests.conftest import *
  from ray.data.tests.test_partitioning import PathPartitionEncoder
  from ray.tests.conftest import *
label_desc: Use image node_id 10 for calling a global function with example usage: test_default_file_metadata_provider_many_files_diff_dirs(ray_start_regular, propagate_logs, caplog, fs, data_path, endpoint_url) without return types