def METHOD_NAME(
    source_path: pathlib.Path,
    destination_path: pathlib.Path,
    fs: Optional[AbstractFileSystem] = None,
) -> None:
    """Copy files from one path to another.
    The source path must either be an existing file or folder. If the source is a folder, the destination path is
    interpreted as a folder as well. If the source is a file, the destination path is interpreted as a file too.
    Files in a folder are copied recursively and efficiently using multiple threads.
    """
    if fs is None:
        fs = _filesystem()
    def _copy(from_path: pathlib.Path, to_path: pathlib.Path) -> Optional[Exception]:
        _logger.debug(f"Copying {str(from_path)} -> {str(to_path)}")
        try:
            # NOTE: S3 does not have a concept of directories, so we do not need to create one.
            if isinstance(fs, LocalFileSystem):
                fs.makedirs(str(to_path.parent), exist_ok=True)
            fs.put(str(from_path), str(to_path), recursive=False)
        except Exception as ex:
            # Return the exception so that it can be handled in the main thread
            return ex
    # NOTE: Cannot use `S3FileSystem.put(recursive=True)` because it tries to access parent directories
    #       which it does not have access to.
    if source_path.is_dir():
        src = [file for file in source_path.rglob("*") if file.is_file()]
        dst = [destination_path / file.relative_to(source_path) for file in src]
        with concurrent.futures.ThreadPoolExecutor(num_workers) as executor:
            results = executor.map(_copy, src, dst)
        # Raise the first exception found
        exception = next((e for e in results if isinstance(e, Exception)), None)
        if exception:
            raise exception
    else:
        if isinstance(fs, LocalFileSystem):
            fs.makedirs(str(destination_path.parent), exist_ok=True)
        fs.put(str(source_path), str(destination_path))