FEAT-#3829: Add support of storage_options for read_csv_glob (#3830)

Signed-off-by: Alexey Prutskov <alexey.prutskov@intel.com>

modin/core/storage_formats/pandas/parsers.py

@@ -308,11 +308,13 @@ def parse(chunks, **kwargs):
         num_splits = kwargs.pop("num_splits", None)
         index_col = kwargs.get("index_col", None)
 
+        # pop `compression` from kwargs because `bio` below is uncompressed
+        compression = kwargs.pop("compression", "infer")
+        storage_options = kwargs.pop("storage_options", None) or {}
         pandas_dfs = []
         for fname, start, end in chunks:
             if start is not None and end is not None:
-                # pop "compression" from kwargs because bio is uncompressed
-                with OpenFile(fname, "rb", kwargs.pop("compression", "infer")) as bio:
+                with OpenFile(fname, "rb", compression, **storage_options) as bio:
                     if kwargs.get("encoding", None) is not None:
                         header = b"" + bio.readline()
                     else:
@@ -322,7 +324,12 @@ def parse(chunks, **kwargs):
                 pandas_dfs.append(pandas.read_csv(BytesIO(to_read), **kwargs))
             else:
                 # This only happens when we are reading with only one worker (Default)
-                return pandas.read_csv(fname, **kwargs)
+                return pandas.read_csv(
+                    fname,
+                    compression=compression,
+                    storage_options=storage_options,
+                    **kwargs,
+                )
 
         # Combine read in data.
         if len(pandas_dfs) > 1:

modin/experimental/pandas/test/test_io_exp.py

@@ -16,6 +16,7 @@
 import pytest
 import modin.experimental.pandas as pd
 from modin.config import Engine
+from modin.utils import get_current_execution
 from modin.pandas.test.utils import df_equals, teardown_test_files, test_data
 
 
@@ -136,6 +137,31 @@ def test_read_multiple_csv_s3():
     df_equals(modin_df, pandas_df)
 
 
+@pytest.mark.skipif(
+    get_current_execution() != "ExperimentalPandasOnRay",
+    reason=f"Execution {get_current_execution()} isn't supported.",
+)
+@pytest.mark.parametrize(
+    "storage_options",
+    [{"anon": False}, {"anon": True}, {"key": "123", "secret": "123"}, None],
+)
+def test_read_multiple_csv_s3_storage_opts(storage_options):
+    path = "s3://modin-datasets/testing/multiple_csv/"
+    # Test the fact of handling of `storage_options`
+    modin_df = pd.read_csv_glob(path, storage_options=storage_options)
+    pandas_df = pd.concat(
+        [
+            pandas.read_csv(
+                f"{path}test_data{i}.csv",
+                storage_options=storage_options,
+            )
+            for i in range(2)
+        ],
+    ).reset_index(drop=True)
+
+    df_equals(modin_df, pandas_df)
+
+
 @pytest.mark.skipif(
     not Engine.get() == "Ray",
     reason=f"{Engine.get()} does not have experimental API",