Repository: modin-project/modin
Pull request: FEAT-#4766: Support fsspec URLs in `read_csv` and `read_csv_glob`
Signed-off-by: Karthik Velayutham <vkarthik@ponder.io>

<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

This PR adds proper support for fsspec URLs that matches pandas behavior for `read_csv`. 

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
- [x] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #4766 
- [x] tests added and passing
- [x] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->
- [x] added (Issue Number: PR title (PR Number)) and github username to release notes for next major release <!-- e.g. DOCS-#4077: Add release notes template to docs folder (#4078) -->


Problem statement:
FEAT: Add support for Distributed loading on GCS buckets
## Context
The company I work for, heavily uses pandas and Google Cloud Storage (GCS) buckets. We were interested in using Modin to improve performance of loading large datasets and improve data processing tasks. 

## Problem
However, it seems as though using GCS bucket urls for the `read_csv()` function of modin, is quite slow. The GCS bucket file patterns are `gs://`. Loading from local memory or S3 buckets works quite well, but the lack of support for GCS buckets results in Modin to default to pandas, which ends up being slower than using the vanilla pandas library!

## Feature Request
We are requesting support for GCS buckets for distributed loading so that we aren't limited to S3 or local file storage only especially since the company I work for, uses Google Suite products quite a lot and does not use Amazon offerings.

Hints from issue discussion:
Connected with #4742 
@Anando304 thanks for opening the issue! This is a good point and we haven't properly addressed it yet (for reasons I don't know exactly). I'll work on a PR to add support for `gs://`.

Commit message: Add release notes

Patch to review:
diff --git a/docs/release_notes/release_notes-0.16.0.rst b/docs/release_notes/release_notes-0.16.0.rst
index 8a77e2d6af6..241ad20d076 100644
--- a/docs/release_notes/release_notes-0.16.0.rst
+++ b/docs/release_notes/release_notes-0.16.0.rst
@@ -106,6 +106,7 @@ Key Features and Updates
   * FEAT-#4664: Finalize compatibility support for Python 3.6 (#4800)
   * FEAT-#4746: Sync interchange protocol with recent API changes (#4763)
   * FEAT-#4733: Support fastparquet as engine for `read_parquet` (#4807)
+  * FEAT-#4766: Support fsspec URLs in `read_csv` and `read_csv_glob` (#4898)
 
 Contributors
 ------------
diff --git a/modin/core/io/file_dispatcher.py b/modin/core/io/file_dispatcher.py
index 94c1dbb354c..955cddeb939 100644
--- a/modin/core/io/file_dispatcher.py
+++ b/modin/core/io/file_dispatcher.py
@@ -23,9 +23,9 @@
 from modin.config import StorageFormat
 from modin.logging import ClassLogger
 import numpy as np
+from pandas.io.common import is_url, is_fsspec_url
 
 NOT_IMPLEMENTED_MESSAGE = "Implement in children classes!"
-_SUPPORTED_PROTOCOLS = {"s3", "S3", "http", "https"}
 
 
 class OpenFile:
@@ -208,12 +208,10 @@ def get_path(cls, file_path):
 
         Notes
         -----
-        if `file_path` is an S3 bucket, parameter will be returned as is, otherwise
+        if `file_path` is a URL, parameter will be returned as is, otherwise
         absolute path will be returned.
         """
-        if isinstance(file_path, str) and (
-            fsspec.core.split_protocol(file_path)[0] in _SUPPORTED_PROTOCOLS
-        ):
+        if is_fsspec_url(file_path) or is_url(file_path):
             return file_path
         else:
             return os.path.abspath(file_path)
@@ -257,15 +255,9 @@ def file_exists(cls, file_path, storage_options=None):
         bool
             Whether file exists or not.
         """
-        if (
-            not isinstance(file_path, str)
-            or fsspec.core.split_protocol(file_path)[0] not in _SUPPORTED_PROTOCOLS
-        ):
+        if not is_fsspec_url(file_path) and not is_url(file_path):
             return os.path.exists(file_path)
 
-        # `file_path` may start with a capital letter, which isn't supported by `fsspec.core.url_to_fs` used below.
-        file_path = file_path[0].lower() + file_path[1:]
-
         from botocore.exceptions import (
             NoCredentialsError,
             EndpointConnectionError,
diff --git a/modin/core/io/text/csv_glob_dispatcher.py b/modin/core/io/text/csv_glob_dispatcher.py
index aae849bcead..f8f1dc13b51 100644
--- a/modin/core/io/text/csv_glob_dispatcher.py
+++ b/modin/core/io/text/csv_glob_dispatcher.py
@@ -24,14 +24,12 @@
 
 import pandas
 import pandas._libs.lib as lib
-from pandas.io.common import is_url
+from pandas.io.common import is_url, is_fsspec_url, stringify_path
 
 from modin.config import NPartitions
 from modin.core.io.file_dispatcher import OpenFile
 from modin.core.io.text.csv_dispatcher import CSVDispatcher
 
-_SUPPORTED_PROTOCOLS = {"s3", "S3"}
-
 
 class CSVGlobDispatcher(CSVDispatcher):
     """Class contains utils for reading multiple `.csv` files simultaneously."""
@@ -54,7 +52,7 @@ def _read(cls, filepath_or_buffer, **kwargs):
             Query compiler with imported data for further processing.
         """
         # Ensures that the file is a string file path. Otherwise, default to pandas.
-        filepath_or_buffer = cls.get_path_or_buffer(filepath_or_buffer)
+        filepath_or_buffer = cls.get_path_or_buffer(stringify_path(filepath_or_buffer))
         if isinstance(filepath_or_buffer, str):
             # os.altsep == None on Linux
             is_folder = any(
@@ -331,18 +329,9 @@ def file_exists(cls, file_path: str, storage_options=None) -> bool:
         bool
             True if the path is valid.
         """
-        if isinstance(file_path, str) and is_url(file_path):
-            raise NotImplementedError("`read_csv_glob` supports only s3-like paths.")
-
-        if (
-            not isinstance(file_path, str)
-            or fsspec.core.split_protocol(file_path)[0] not in _SUPPORTED_PROTOCOLS
-        ):
+        if not is_fsspec_url(file_path) and not is_url(file_path):
             return len(glob.glob(file_path)) > 0
 
-        # `file_path` may start with a capital letter, which isn't supported by `fsspec.core.url_to_fs` used below.
-        file_path = file_path[0].lower() + file_path[1:]
-
         from botocore.exceptions import (
             NoCredentialsError,
             EndpointConnectionError,
@@ -384,14 +373,11 @@ def get_path(cls, file_path: str) -> list:
         list
             List of strings of absolute file paths.
         """
-        if fsspec.core.split_protocol(file_path)[0] not in _SUPPORTED_PROTOCOLS:
+        if not is_fsspec_url(file_path) and not is_url(file_path):
             relative_paths = glob.glob(file_path)
             abs_paths = [os.path.abspath(path) for path in relative_paths]
             return abs_paths
 
-        # `file_path` may start with a capital letter, which isn't supported by `fsspec.core.url_to_fs` used below.
-        file_path = file_path[0].lower() + file_path[1:]
-
         from botocore.exceptions import (
             NoCredentialsError,
             EndpointConnectionError,
@@ -402,8 +388,8 @@ def get_file_path(fs_handle) -> List[str]:
             file_paths = fs_handle.glob(file_path)
             if len(file_paths) == 0 and not fs_handle.exists(file_path):
                 raise FileNotFoundError(f"Path <{file_path}> isn't available.")
-            s3_addresses = [f"s3://{path}" for path in file_paths]
-            return s3_addresses
+            fs_addresses = [fs_handle.unstrip_protocol(path) for path in file_paths]
+            return fs_addresses
 
         fs, _ = fsspec.core.url_to_fs(file_path)
         try:
diff --git a/modin/core/io/text/text_file_dispatcher.py b/modin/core/io/text/text_file_dispatcher.py
index eb204b81d2e..7a329fed5ea 100644
--- a/modin/core/io/text/text_file_dispatcher.py
+++ b/modin/core/io/text/text_file_dispatcher.py
@@ -28,6 +28,7 @@
 import pandas
 import pandas._libs.lib as lib
 from pandas.core.dtypes.common import is_list_like
+from pandas.io.common import stringify_path
 
 from modin.core.io.file_dispatcher import FileDispatcher, OpenFile
 from modin.core.storage_formats.pandas.utils import compute_chunksize
@@ -963,6 +964,7 @@ def _read(cls, filepath_or_buffer, **kwargs):
         new_query_compiler : BaseQueryCompiler
             Query compiler with imported data for further processing.
         """
+        filepath_or_buffer = stringify_path(filepath_or_buffer)
         filepath_or_buffer_md = (
             cls.get_path(filepath_or_buffer)
             if isinstance(filepath_or_buffer, str)
diff --git a/modin/experimental/pandas/test/test_io_exp.py b/modin/experimental/pandas/test/test_io_exp.py
index e10dc9fe6ec..766bc0398c1 100644
--- a/modin/experimental/pandas/test/test_io_exp.py
+++ b/modin/experimental/pandas/test/test_io_exp.py
@@ -168,7 +168,7 @@ def test_read_single_csv_with_parse_dates(self, parse_dates):
     Engine.get() != "Ray", reason="Currently only support Ray engine for glob paths."
 )
 def test_read_multiple_csv_s3():
-    path = "S3://modin-datasets/testing/multiple_csv/test_data*.csv"
+    path = "s3://modin-datasets/testing/multiple_csv/test_data*.csv"
 
     def _pandas_read_csv_glob(path, storage_options):
         pandas_dfs = [
diff --git a/modin/pandas/test/test_io.py b/modin/pandas/test/test_io.py
index 7614a3c5139..a8fe952d0e8 100644
--- a/modin/pandas/test/test_io.py
+++ b/modin/pandas/test/test_io.py
@@ -915,6 +915,13 @@ def test_read_csv_categories(self):
             dtype={"one": "int64", "two": "category"},
         )
 
+    def test_read_csv_gs(self):
+        eval_io(
+            fn_name="read_csv",
+            # read_csv kwargs
+            filepath_or_buffer="gs://modin-testing/testing/multiple_csv/test_data0.csv",
+        )
+
     @pytest.mark.parametrize("encoding", [None, "utf-8"])
     @pytest.mark.parametrize("encoding_errors", ["strict", "ignore"])
     @pytest.mark.parametrize(


Code review comments:
diff_hunk: @@ -331,18 +329,9 @@ def file_exists(cls, file_path: str, storage_options=None) -> bool:
         bool
             True if the path is valid.
         """
-        if isinstance(file_path, str) and is_url(file_path):
path: modin/core/io/text/csv_glob_dispatcher.py
text: @user1:
What if someone passes in a GitHub raw user content file - with an https irl prefix - would that work?

@author:
Yeah actually I think it shouldn't work for the glob case. I remember Alexey bringing this up, so I'll revert this back.