Repository: modin-project/modin
Issue category: Feature Request

Support fastparquet engine for `read_parquet`
**Is your feature request related to a problem? Please describe.**
Currently our implementation of `read_parquet` only supports PyArrow. Given that we have some more user demand, we should also support fastparquet as a library to read parquet files. 



Functions to edit:
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher.get_fs_and_files
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher._get_dataset
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher._get_fs_and_fs_path
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher.call_deploy
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher.build_index
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher.build_query_compiler
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher._read
modin/core/storage_formats/pandas/parsers.py:PandasParquetParser.parse
modin/core/storage_formats/pandas/parsers.py:PandasParquetParser
Functions added: modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset.__init__
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset.pandas_metadata
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset.columns
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset.engine
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset.files
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset.row_groups_per_file
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset.fs
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset.fs_path
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset.to_pandas_dataframe
modin/core/io/column_stores/parquet_dispatcher.py:ColumnStoreDataset._get_files
modin/core/io/column_stores/parquet_dispatcher.py:PyArrowDataset
modin/core/io/column_stores/parquet_dispatcher.py:PyArrowDataset._init_dataset
modin/core/io/column_stores/parquet_dispatcher.py:PyArrowDataset.pandas_metadata
modin/core/io/column_stores/parquet_dispatcher.py:PyArrowDataset.columns
modin/core/io/column_stores/parquet_dispatcher.py:PyArrowDataset.engine
modin/core/io/column_stores/parquet_dispatcher.py:PyArrowDataset.row_groups_per_file
modin/core/io/column_stores/parquet_dispatcher.py:PyArrowDataset.files
modin/core/io/column_stores/parquet_dispatcher.py:PyArrowDataset.to_pandas_dataframe
modin/core/io/column_stores/parquet_dispatcher.py:FastParquetDataset
modin/core/io/column_stores/parquet_dispatcher.py:FastParquetDataset._init_dataset
modin/core/io/column_stores/parquet_dispatcher.py:FastParquetDataset.pandas_metadata
modin/core/io/column_stores/parquet_dispatcher.py:FastParquetDataset.columns
modin/core/io/column_stores/parquet_dispatcher.py:FastParquetDataset.engine
modin/core/io/column_stores/parquet_dispatcher.py:FastParquetDataset.row_groups_per_file
modin/core/io/column_stores/parquet_dispatcher.py:FastParquetDataset.files
modin/core/io/column_stores/parquet_dispatcher.py:FastParquetDataset.to_pandas_dataframe
modin/core/io/column_stores/parquet_dispatcher.py:FastParquetDataset._get_fastparquet_files
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher
modin/core/io/column_stores/parquet_dispatcher.py:ParquetDispatcher.get_dataset
modin/core/storage_formats/pandas/parsers.py:PandasParquetParser._read_row_group_chunk
modin/core/storage_formats/pandas/parsers.py:PandasParquetParser

Patch:
diff --git a/docs/release_notes/release_notes-0.16.0.rst b/docs/release_notes/release_notes-0.16.0.rst
index 9f5c253a89d..8e29f4206ce 100644
--- a/docs/release_notes/release_notes-0.16.0.rst
+++ b/docs/release_notes/release_notes-0.16.0.rst
@@ -93,6 +93,7 @@ Key Features and Updates
   * FEAT-#4725: Make index and columns lazy in Modin DataFrame (#4726)
   * 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)
 
 Contributors
 ------------
diff --git a/environment-dev.yml b/environment-dev.yml
index 1a23426a22e..a1a8011c839 100644
--- a/environment-dev.yml
+++ b/environment-dev.yml
@@ -38,6 +38,7 @@ dependencies:
   - psycopg2
   - mypy
   - pandas-stubs
+  - fastparquet
   - pip:
       - xgboost>=1.6.0
       - modin-spreadsheet>=0.1.1
diff --git a/modin/core/io/column_stores/parquet_dispatcher.py b/modin/core/io/column_stores/parquet_dispatcher.py
index ea5f9e08186..d68a82dc967 100644
--- a/modin/core/io/column_stores/parquet_dispatcher.py
+++ b/modin/core/io/column_stores/parquet_dispatcher.py
@@ -15,12 +15,11 @@
 
 import os
 
+import json
 import fsspec
-from fsspec.core import split_protocol, url_to_fs
-from fsspec.registry import get_filesystem_class
+from fsspec.core import url_to_fs
 from fsspec.spec import AbstractBufferedFile
 import numpy as np
-from pandas.io.common import is_fsspec_url
 from packaging import version
 
 from modin.core.storage_formats.pandas.utils import compute_chunksize
@@ -28,30 +27,132 @@
 
 
 from modin.core.io.column_stores.column_store_dispatcher import ColumnStoreDispatcher
-from modin.utils import import_optional_dependency
+from modin.utils import import_optional_dependency, _inherit_docstrings
+
+
+class ColumnStoreDataset:
+    """
+    Base class that encapsulates Parquet engine-specific details.
+
+    This class exposes a set of functions that are commonly used in the
+    `read_parquet` implementation.
+
+    Attributes
+    ----------
+    path : str, path object or file-like object
+        The filepath of the parquet file in local filesystem or hdfs.
+    storage_options : dict
+        Parameters for specific storage engine.
+    _fs_path : str, path object or file-like object
+        The filepath or handle of the parquet dataset specific to the
+        filesystem implementation. E.g. for `s3://test/example`, _fs
+        would be set to S3FileSystem and _fs_path would be `test/example`.
+    _fs : Filesystem
+        Filesystem object specific to the given parquet file/dataset.
+    dataset : ParquetDataset or ParquetFile
+        Underlying dataset implementation for PyArrow and fastparquet
+        respectively.
+    _row_groups_per_file : list
+        List that contains the number of row groups for each file in the
+        given parquet dataset.
+    _files : list
+        List that contains the full paths of the parquet files in the dataset.
+    """
+
+    def __init__(self, path, storage_options):  # noqa : PR01
+        self.path = path
+        self.storage_options = storage_options
+        self._fs_path = None
+        self._fs = None
+        self.dataset = self._init_dataset()
+        self._row_groups_per_file = None
+        self._files = None
+
+    @property
+    def pandas_metadata(self):
+        """Return the pandas metadata of the dataset."""
+        raise NotImplementedError
+
+    @property
+    def columns(self):
+        """Return the list of columns in the dataset."""
+        raise NotImplementedError
+
+    @property
+    def engine(self):
+        """Return string representing what engine is being used."""
+        raise NotImplementedError
+
+    # TODO: make this cache_readonly after docstring inheritance is fixed.
+    @property
+    def files(self):
+        """Return the list of formatted file paths of the dataset."""
+        raise NotImplementedError
+
+    # TODO: make this cache_readonly after docstring inheritance is fixed.
+    @property
+    def row_groups_per_file(self):
+        """Return a list with the number of row groups per file."""
+        raise NotImplementedError
+
+    @property
+    def fs(self):
+        """
+        Return the filesystem object associated with the dataset path.
 
+        Returns
+        -------
+        filesystem
+            Filesystem object.
+        """
+        if self._fs is None:
+            if isinstance(self.path, AbstractBufferedFile):
+                self._fs = self.path.fs
+            else:
+                self._fs, self._fs_path = url_to_fs(self.path, **self.storage_options)
+        return self._fs
 
-class ParquetDispatcher(ColumnStoreDispatcher):
-    """Class handles utils for reading `.parquet` files."""
+    @property
+    def fs_path(self):
+        """
+        Return the filesystem-specific path or file handle.
 
-    @classmethod
-    def get_fs_and_files(cls, path, storage_options):
+        Returns
+        -------
+        fs_path : str, path object or file-like object
+            String path specific to filesystem or a file handle.
+        """
+        if self._fs_path is None:
+            if isinstance(self.path, AbstractBufferedFile):
+                self._fs_path = self.path
+            else:
+                self._fs, self._fs_path = url_to_fs(self.path, **self.storage_options)
+        return self._fs_path
+
+    def to_pandas_dataframe(self, columns):
         """
-        Retrieve filesystem interface and list of files from path.
+        Read the given columns as a pandas dataframe.
 
         Parameters
         ----------
-        path : str, path object or file-like object
-            Path to dataset.
-        storage_options : dict
-            Parameters for specific storage engine.
+        columns : list
+            List of columns that should be read from file.
+        """
+        raise NotImplementedError
+
+    def _get_files(self, files):
+        """
+        Retrieve list of formatted file names in dataset path.
+
+        Parameters
+        ----------
+        files : list
+            List of files from path.
 
         Returns
         -------
-        filesystem: Any
-            Protocol implementation of registry.
-        files: list
-            List of files from path.
+        fs_files : list
+            List of files from path with fs-protocol prepended.
         """
         # Older versions of fsspec doesn't support unstrip_protocol(). It
         # was only added relatively recently:
@@ -63,73 +164,187 @@ def _unstrip_protocol(protocol, path):
                     return path
             return f"{protos[0]}://{path}"
 
-        if isinstance(path, AbstractBufferedFile):
-            return path.fs, [path]
-        fs = get_filesystem_class(split_protocol(path)[0])(**storage_options)
-        dataset = cls._get_dataset(path, storage_options)
+        if isinstance(self.path, AbstractBufferedFile):
+            return [self.path]
         # version.parse() is expensive, so we can split this into two separate loops
         if version.parse(fsspec.__version__) < version.parse("2022.5.0"):
-            files = [_unstrip_protocol(fs.protocol, fpath) for fpath in dataset.files]
+            fs_files = [_unstrip_protocol(self.fs.protocol, fpath) for fpath in files]
         else:
-            files = [fs.unstrip_protocol(fpath) for fpath in dataset.files]
+            fs_files = [self.fs.unstrip_protocol(fpath) for fpath in files]
 
-        return fs, files
+        return fs_files
 
-    @classmethod
-    def _get_dataset(cls, path, storage_options):
-        """
-        Retrieve parquet dataset for given path.
 
-        Parameters
-        ----------
-        path : str, path object or file-like object
-            Path to dataset.
-        storage_options : dict
-            Parameters for specific storage engine.
-
-        Returns
-        -------
-        dataset : ParquetDataset
-            Dataset object of given Parquet file-path.
-        """
+@_inherit_docstrings(ColumnStoreDataset)
+class PyArrowDataset(ColumnStoreDataset):
+    def _init_dataset(self):  # noqa: GL08
         from pyarrow.parquet import ParquetDataset
 
-        fs, fs_path = cls._get_fs_and_fs_path(path, storage_options)
-        dataset = ParquetDataset(fs_path, filesystem=fs, use_legacy_dataset=False)
-        return dataset
+        return ParquetDataset(
+            self.fs_path, filesystem=self.fs, use_legacy_dataset=False
+        )
+
+    @property
+    def pandas_metadata(self):
+        return self.dataset.schema.pandas_metadata
+
+    @property
+    def columns(self):
+        return self.dataset.schema.names
+
+    @property
+    def engine(self):
+        return "pyarrow"
+
+    @property
+    def row_groups_per_file(self):
+        from pyarrow.parquet import ParquetFile
+
+        if self._row_groups_per_file is None:
+            row_groups_per_file = []
+            # Count up the total number of row groups across all files and
+            # keep track of row groups per file to use later.
+            for file in self.files:
+                with self.fs.open(file) as f:
+                    row_groups = ParquetFile(f).num_row_groups
+                    row_groups_per_file.append(row_groups)
+            self._row_groups_per_file = row_groups_per_file
+        return self._row_groups_per_file
+
+    @property
+    def files(self):
+        if self._files is None:
+            self._files = self._get_files(self.dataset.files)
+        return self._files
+
+    def to_pandas_dataframe(
+        self,
+        columns,
+    ):
+        from pyarrow.parquet import read_table
+
+        return read_table(self.path, columns=columns, filesystem=self.fs).to_pandas()
+
+
+@_inherit_docstrings(ColumnStoreDataset)
+class FastParquetDataset(ColumnStoreDataset):
+    def _init_dataset(self):  # noqa: GL08
+        from fastparquet import ParquetFile
+
+        return ParquetFile(self.fs_path, fs=self.fs)
+
+    @property
+    def pandas_metadata(self):
+        if "pandas" not in self.dataset.key_value_metadata:
+            return {}
+        return json.loads(self.dataset.key_value_metadata["pandas"])
+
+    @property
+    def columns(self):
+        return self.dataset.columns
+
+    @property
+    def engine(self):
+        return "fastparquet"
+
+    @property
+    def row_groups_per_file(self):
+        from fastparquet import ParquetFile
+
+        if self._row_groups_per_file is None:
+            row_groups_per_file = []
+            # Count up the total number of row groups across all files and
+            # keep track of row groups per file to use later.
+            for file in self.files:
+                with self.fs.open(file) as f:
+                    row_groups = ParquetFile(f).info["row_groups"]
+                    row_groups_per_file.append(row_groups)
+            self._row_groups_per_file = row_groups_per_file
+        return self._row_groups_per_file
+
+    @property
+    def files(self):
+        if self._files is None:
+            self._files = self._get_files(self._get_fastparquet_files())
+        return self._files
+
+    def to_pandas_dataframe(self, columns):
+        return self.dataset.to_pandas(columns=columns)
+
+    def _get_fastparquet_files(self):  # noqa: GL08
+        # fastparquet doesn't have a nice method like PyArrow, so we
+        # have to copy some of their logic here while we work on getting
+        # an easier method to get a list of valid files.
+        # See: https://github.com/dask/fastparquet/issues/795
+        if "*" in self.path:
+            files = self.fs.glob(self.path)
+        else:
+            files = [
+                f
+                for f in self.fs.find(self.path)
+                if f.endswith(".parquet") or f.endswith(".parq")
+            ]
+        return files
+
+
+class ParquetDispatcher(ColumnStoreDispatcher):
+    """Class handles utils for reading `.parquet` files."""
 
     @classmethod
-    def _get_fs_and_fs_path(cls, path, storage_options):
+    def get_dataset(cls, path, engine, storage_options):
         """
-        Retrieve filesystem interface and filesystem-specific path.
+        Retrieve Parquet engine specific Dataset implementation.
 
         Parameters
         ----------
         path : str, path object or file-like object
-            Path to dataset.
+            The filepath of the parquet file in local filesystem or hdfs.
+        engine : str
+            Parquet library to use (only 'PyArrow' is supported for now).
         storage_options : dict
             Parameters for specific storage engine.
 
         Returns
         -------
-        filesystem : Any
-            Protocol implementation of registry.
-        fs_path : list
-            Filesystem's specific path.
+        Dataset
+            Either a PyArrowDataset or FastParquetDataset object.
         """
-        return (
-            url_to_fs(path, **storage_options) if is_fsspec_url(path) else (None, path)
-        )
+        if engine == "auto":
+            # We follow in concordance with pandas
+            engine_classes = [PyArrowDataset, FastParquetDataset]
+
+            error_msgs = ""
+            for engine_class in engine_classes:
+                try:
+                    return engine_class(path, storage_options)
+                except ImportError as err:
+                    error_msgs += "\n - " + str(err)
+
+            raise ImportError(
+                "Unable to find a usable engine; "
+                + "tried using: 'pyarrow', 'fastparquet'.\n"
+                + "A suitable version of "
+                + "pyarrow or fastparquet is required for parquet "
+                + "support.\n"
+                + "Trying to import the above resulted in these errors:"
+                + f"{error_msgs}"
+            )
+        elif engine == "pyarrow":
+            return PyArrowDataset(path, storage_options)
+        elif engine == "fastparquet":
+            return FastParquetDataset(path, storage_options)
+        else:
+            raise ValueError("engine must be one of 'pyarrow', 'fastparquet'")
 
     @classmethod
-    def call_deploy(cls, path, col_partitions, storage_options, **kwargs):
+    def call_deploy(cls, dataset, col_partitions, storage_options, **kwargs):
         """
         Deploy remote tasks to the workers with passed parameters.
 
         Parameters
         ----------
-        path : str, path object or file-like object
-            Name of the file to read.
+        dataset : Dataset
+            Dataset object of Parquet file/files.
         col_partitions : list
             List of arrays with columns names that should be read
             by each partition.
@@ -143,7 +358,6 @@ def call_deploy(cls, path, col_partitions, storage_options, **kwargs):
         List
             Array with references to the task deploy result for each partition.
         """
-        from pyarrow.parquet import ParquetFile
         from modin.core.storage_formats.pandas.parsers import ParquetFileToRead
 
         # If we don't have any columns to read, we should just return an empty
@@ -151,17 +365,9 @@ def call_deploy(cls, path, col_partitions, storage_options, **kwargs):
         if len(col_partitions) == 0:
             return []
 
-        filesystem, parquet_files = cls.get_fs_and_files(path, storage_options)
-
-        row_groups_per_file = []
-        num_row_groups = 0
-        # Count up the total number of row groups across all files and
-        # keep track of row groups per file to use later.
-        for file in parquet_files:
-            with filesystem.open(file) as f:
-                row_groups = ParquetFile(f).num_row_groups
-                row_groups_per_file.append(row_groups)
-                num_row_groups += row_groups
+        row_groups_per_file = dataset.row_groups_per_file
+        num_row_groups = sum(row_groups_per_file)
+        parquet_files = dataset.files
 
         # step determines how many row groups are going to be in a partition
         step = compute_chunksize(
@@ -221,6 +427,7 @@ def call_deploy(cls, path, col_partitions, storage_options, **kwargs):
                         cls.parse,
                         files_for_parser=files_to_read,
                         columns=cols,
+                        engine=dataset.engine,
                         num_returns=3,
                         storage_options=storage_options,
                         **kwargs,
@@ -269,20 +476,18 @@ def build_partition(cls, partition_ids, column_widths):
         )
 
     @classmethod
-    def build_index(cls, path, partition_ids, index_columns, storage_options):
+    def build_index(cls, dataset, partition_ids, index_columns):
         """
         Compute index and its split sizes of resulting Modin DataFrame.
 
         Parameters
         ----------
-        path : Pathlike
-            Path to dataset.
+        dataset : Dataset
+            Dataset object of Parquet file/files.
         partition_ids : list
             Array with references to the partitions data.
         index_columns : list
             List of index columns specified by pandas metadata.
-        storage_options : dict
-            Parameters for specific storage engine.
 
         Returns
         -------
@@ -297,8 +502,6 @@ def build_index(cls, path, partition_ids, index_columns, storage_options):
         -----
         See `build_partition` for more detail on the contents of partitions_ids.
         """
-        from pyarrow.parquet import read_table
-
         range_index = True
         column_names_to_read = []
         for column in index_columns:
@@ -314,12 +517,9 @@ def build_index(cls, path, partition_ids, index_columns, storage_options):
         # For the second check, let us consider the case where we have an empty dataframe,
         # that has a valid index.
         if range_index or (len(partition_ids) == 0 and len(column_names_to_read) != 0):
-            fs, fs_path = cls._get_fs_and_fs_path(path, storage_options)
-            complete_index = (
-                read_table(fs_path, columns=column_names_to_read, filesystem=fs)
-                .to_pandas()
-                .index
-            )
+            complete_index = dataset.to_pandas_dataframe(
+                columns=column_names_to_read
+            ).index
         # Empty DataFrame case
         elif len(partition_ids) == 0:
             return [], False
@@ -330,14 +530,14 @@ def build_index(cls, path, partition_ids, index_columns, storage_options):
         return complete_index, range_index or (len(index_columns) == 0)
 
     @classmethod
-    def build_query_compiler(cls, path, columns, index_columns, **kwargs):
+    def build_query_compiler(cls, dataset, columns, index_columns, **kwargs):
         """
         Build query compiler from deployed tasks outputs.
 
         Parameters
         ----------
-        path : str, path object or file-like object
-            Path to the file to read.
+        dataset : Dataset
+            Dataset object of Parquet file/files.
         columns : list
             List of columns that should be read from file.
         index_columns : list
@@ -352,10 +552,10 @@ def build_query_compiler(cls, path, columns, index_columns, **kwargs):
         """
         storage_options = kwargs.pop("storage_options", {}) or {}
         col_partitions, column_widths = cls.build_columns(columns)
-        partition_ids = cls.call_deploy(path, col_partitions, storage_options, **kwargs)
-        index, sync_index = cls.build_index(
-            path, partition_ids, index_columns, storage_options
+        partition_ids = cls.call_deploy(
+            dataset, col_partitions, storage_options, **kwargs
         )
+        index, sync_index = cls.build_index(dataset, partition_ids, index_columns)
         remote_parts = cls.build_partition(partition_ids, column_widths)
         if len(partition_ids) > 0:
             row_lengths = [part.length() for part in remote_parts.T[0]]
@@ -382,8 +582,8 @@ def _read(cls, path, engine, columns, **kwargs):
         ----------
         path : str, path object or file-like object
             The filepath of the parquet file in local filesystem or hdfs.
-        engine : str
-            Parquet library to use (only 'PyArrow' is supported for now).
+        engine : {"auto", "pyarrow", "fastparquet"}
+            Parquet library to use.
         columns : list
             If not None, only these columns will be read from the file.
         **kwargs : dict
@@ -428,18 +628,19 @@ def _read(cls, path, engine, columns, **kwargs):
                     reason="Mixed partitioning columns in Parquet",
                     **kwargs,
                 )
-        dataset = cls._get_dataset(path, kwargs.get("storage_options") or {})
+
+        dataset = cls.get_dataset(path, engine, kwargs.get("storage_options") or {})
         index_columns = (
-            dataset.schema.pandas_metadata.get("index_columns", [])
-            if dataset.schema.pandas_metadata
+            dataset.pandas_metadata.get("index_columns", [])
+            if dataset.pandas_metadata
             else []
         )
         # If we have columns as None, then we default to reading in all the columns
-        column_names = dataset.schema.names if not columns else columns
+        column_names = columns if columns else dataset.columns
         columns = [
             c
             for c in column_names
             if c not in index_columns and not PQ_INDEX_REGEX.match(c)
         ]
 
-        return cls.build_query_compiler(path, columns, index_columns, **kwargs)
+        return cls.build_query_compiler(dataset, columns, index_columns, **kwargs)
diff --git a/modin/core/storage_formats/pandas/parsers.py b/modin/core/storage_formats/pandas/parsers.py
index d1fb2fb189b..e08f4da8451 100644
--- a/modin/core/storage_formats/pandas/parsers.py
+++ b/modin/core/storage_formats/pandas/parsers.py
@@ -663,23 +663,53 @@ class ParquetFileToRead(NamedTuple):
 
 @doc(_doc_pandas_parser_class, data_type="PARQUET data")
 class PandasParquetParser(PandasParser):
+    @staticmethod
+    def _read_row_group_chunk(
+        f, row_group_start, row_group_end, columns, engine
+    ):  # noqa: GL08
+        if engine == "pyarrow":
+            from pyarrow.parquet import ParquetFile
+
+            return (
+                ParquetFile(f)
+                .read_row_groups(
+                    range(
+                        row_group_start,
+                        row_group_end,
+                    ),
+                    columns=columns,
+                    use_pandas_metadata=True,
+                )
+                .to_pandas()
+            )
+        elif engine == "fastparquet":
+            from fastparquet import ParquetFile
+
+            return ParquetFile(f)[row_group_start:row_group_end].to_pandas(
+                columns=columns
+            )
+        else:
+            # We shouldn't ever come to this case, so something went wrong
+            raise ValueError(
+                f"engine must be one of 'pyarrow', 'fastparquet', got: {engine}"
+            )
+
     @staticmethod
     @doc(
         _doc_parse_func,
         parameters="""files_for_parser : list
     List of files to be read.
+engine : str
+    Parquet library to use (either PyArrow or fastparquet).
 """,
     )
-    def parse(files_for_parser, **kwargs):
-        from pyarrow.parquet import ParquetFile
-
+    def parse(files_for_parser, engine, **kwargs):
         columns = kwargs.get("columns", None)
         storage_options = kwargs.pop("storage_options", {}) or {}
-
         chunks = []
         # `single_worker_read` just passes in a string path
         if isinstance(files_for_parser, str):
-            return pandas.read_parquet(files_for_parser, **kwargs)
+            return pandas.read_parquet(files_for_parser, engine=engine, **kwargs)
 
         for file_for_parser in files_for_parser:
             if isinstance(file_for_parser.path, IOBase):
@@ -687,19 +717,12 @@ def parse(files_for_parser, **kwargs):
             else:
                 context = fsspec.open(file_for_parser.path, **storage_options)
             with context as f:
-                chunk = (
-                    ParquetFile(f)
-                    .read_row_groups(
-                        list(
-                            range(
-                                file_for_parser.row_group_start,
-                                file_for_parser.row_group_end,
-                            )
-                        ),
-                        columns=columns,
-                        use_pandas_metadata=True,
-                    )
-                    .to_pandas()
+                chunk = PandasParquetParser._read_row_group_chunk(
+                    f,
+                    file_for_parser.row_group_start,
+                    file_for_parser.row_group_end,
+                    columns,
+                    engine,
                 )
             chunks.append(chunk)
         df = pandas.concat(chunks)
diff --git a/requirements-dev.txt b/requirements-dev.txt
index cf89d697ed1..b95f18784c2 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -44,3 +44,4 @@ numpydoc==1.1.0
 fuzzydata>=0.0.6
 mypy
 pandas-stubs
+fastparquet
\ No newline at end of file
diff --git a/requirements/environment-py36.yml b/requirements/environment-py36.yml
index e82fa3c5cb7..8986154994f 100644
--- a/requirements/environment-py36.yml
+++ b/requirements/environment-py36.yml
@@ -38,6 +38,7 @@ dependencies:
   - pymssql
   - psycopg2
   - jupyter>=1.0.0 # modin-spreadsheet needs that, but it won't install on Windows on Py36 via pip
+  - fastparquet
   - pip:
       - modin-spreadsheet>=0.1.1
       - tqdm
diff --git a/requirements/requirements-py36.txt b/requirements/requirements-py36.txt
index a6f1f5e8322..8bba9e29d1a 100644
--- a/requirements/requirements-py36.txt
+++ b/requirements/requirements-py36.txt
@@ -25,3 +25,4 @@ scikit-learn
 pickle5
 tqdm
 modin-spreadsheet>=0.1.1
+fastparquet
\ No newline at end of file

Test patch: diff --git a/modin/pandas/test/test_io.py b/modin/pandas/test/test_io.py
index a664fd12c6c..3b5a38450ed 100644
--- a/modin/pandas/test/test_io.py
+++ b/modin/pandas/test/test_io.py
@@ -1346,17 +1346,19 @@ def test_read_table_empty_frame(self, make_csv_file):
 class TestParquet:
     @pytest.mark.parametrize("columns", [None, ["col1"]])
     @pytest.mark.parametrize("row_group_size", [None, 100, 1000, 10_000])
+    @pytest.mark.parametrize("engine", ["auto", "pyarrow", "fastparquet"])
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",
         reason="The reason of tests fail in `cloud` mode is unknown for now - issue #3264",
     )
-    def test_read_parquet(self, make_parquet_file, columns, row_group_size):
+    def test_read_parquet(self, make_parquet_file, columns, row_group_size, engine):
         unique_filename = get_unique_filename(extension="parquet")
         make_parquet_file(filename=unique_filename, row_group_size=row_group_size)
 
         eval_io(
             fn_name="read_parquet",
             # read_parquet kwargs
+            engine=engine,
             path=unique_filename,
             columns=columns,
         )
@@ -1385,12 +1387,13 @@ def test_read_parquet_indexing_by_column(self, make_parquet_file):
     @pytest.mark.parametrize(
         "rows_per_file", [[1000] * 40, [0, 0, 40_000], [10_000, 10_000] + [100] * 200]
     )
+    @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",
         reason="The reason of tests fail in `cloud` mode is unknown for now - issue #3264",
     )
     def test_read_parquet_directory(
-        self, make_parquet_dir, columns, row_group_size, rows_per_file
+        self, make_parquet_dir, columns, row_group_size, rows_per_file, engine
     ):
         num_cols = DATASET_SIZE_DICT.get(
             TestDatasetSize.get(), DATASET_SIZE_DICT["Small"]
@@ -1415,31 +1418,37 @@ def test_read_parquet_directory(
         eval_io(
             fn_name="read_parquet",
             # read_parquet kwargs
+            engine=engine,
             path=path,
             columns=columns,
         )
 
+    @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
     @pytest.mark.parametrize("columns", [None, ["col1"]])
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",
         reason="The reason of tests fail in `cloud` mode is unknown for now - issue #3264",
     )
-    def test_read_parquet_partitioned_directory(self, make_parquet_file, columns):
+    def test_read_parquet_partitioned_directory(
+        self, make_parquet_file, columns, engine
+    ):
         unique_filename = get_unique_filename(extension=None)
         make_parquet_file(filename=unique_filename, partitioned_columns=["col1"])
 
         eval_io(
             fn_name="read_parquet",
             # read_parquet kwargs
+            engine=engine,
             path=unique_filename,
             columns=columns,
         )
 
+    @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",
         reason="The reason of tests fail in `cloud` mode is unknown for now - issue #3264",
     )
-    def test_read_parquet_pandas_index(self):
+    def test_read_parquet_pandas_index(self, engine):
         # Ensure modin can read parquet files written by pandas with a non-RangeIndex object
         unique_filename = get_unique_filename(extension="parquet")
         pandas_df = pandas.DataFrame(
@@ -1449,7 +1458,6 @@ def test_read_parquet_pandas_index(self):
                 # Can't do interval index right now because of this bug fix that is planned
                 # to be apart of the pandas 1.5.0 release: https://github.com/pandas-dev/pandas/pull/46034
                 # "idx_interval": pandas.interval_range(start=0, end=2000),
-                "idx_datetime": pandas.date_range(start="1/1/2018", periods=2000),
                 "idx_periodrange": pandas.period_range(
                     start="2017-01-01", periods=2000
                 ),
@@ -1466,28 +1474,42 @@ def test_read_parquet_pandas_index(self):
                 start="1 day", periods=2000
             )
 
+        # There is a non-deterministic bug in the fastparquet engine when we
+        # try to set the index to the datetime column. Please see:
+        # https://github.com/dask/fastparquet/issues/796
+        if engine == "pyarrow":
+            pandas_df["idx_datetime"] = pandas.date_range(
+                start="1/1/2018", periods=2000
+            )
+
         try:
             for col in pandas_df.columns:
                 if col.startswith("idx"):
                     pandas_df.set_index(col).to_parquet(unique_filename)
                     # read the same parquet using modin.pandas
-                    df_equals(
-                        pd.read_parquet(unique_filename),
-                        pandas.read_parquet(unique_filename),
+                    eval_io(
+                        "read_parquet",
+                        # read_parquet kwargs
+                        path=unique_filename,
+                        engine=engine,
                     )
 
             pandas_df.set_index(["idx", "A"]).to_parquet(unique_filename)
-            df_equals(
-                pd.read_parquet(unique_filename), pandas.read_parquet(unique_filename)
+            eval_io(
+                "read_parquet",
+                # read_parquet kwargs
+                path=unique_filename,
+                engine=engine,
             )
         finally:
             os.remove(unique_filename)
 
+    @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",
         reason="The reason of tests fail in `cloud` mode is unknown for now - issue #3264",
     )
-    def test_read_parquet_pandas_index_partitioned(self):
+    def test_read_parquet_pandas_index_partitioned(self, engine):
         # Ensure modin can read parquet files written by pandas with a non-RangeIndex object
         unique_filename = get_unique_filename(extension="parquet")
         pandas_df = pandas.DataFrame(
@@ -1501,48 +1523,60 @@ def test_read_parquet_pandas_index_partitioned(self):
         try:
             pandas_df.set_index("idx").to_parquet(unique_filename, partition_cols=["A"])
             # read the same parquet using modin.pandas
-            df_equals(
-                pd.read_parquet(unique_filename), pandas.read_parquet(unique_filename)
+            eval_io(
+                "read_parquet",
+                # read_parquet kwargs
+                path=unique_filename,
+                engine=engine,
             )
         finally:
             shutil.rmtree(unique_filename)
 
+    @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",
         reason="The reason of tests fail in `cloud` mode is unknown for now - issue #3264",
     )
-    def test_read_parquet_hdfs(self):
+    def test_read_parquet_hdfs(self, engine):
         eval_io(
             fn_name="read_parquet",
             # read_parquet kwargs
             path="modin/pandas/test/data/hdfs.parquet",
+            engine=engine,
         )
 
     @pytest.mark.parametrize("path_type", ["url", "object"])
+    @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",
         reason="The reason of tests fail in `cloud` mode is unknown for now - issue #3264",
     )
-    def test_read_parquet_s3(self, path_type):
+    def test_read_parquet_s3(self, path_type, engine):
         dataset_url = "s3://modin-datasets/testing/test_data.parquet"
         if path_type == "object":
             import s3fs
 
             fs = s3fs.S3FileSystem(anon=True)
             with fs.open(dataset_url, "rb") as file_obj:
-                eval_io("read_parquet", path=file_obj)
+                eval_io("read_parquet", path=file_obj, engine=engine)
         else:
             if PandasCompatVersion.CURRENT == PandasCompatVersion.PY36:
                 pytest.xfail(
                     reason="older pandas.read_parquet does not support storage_options"
                 )
-            eval_io("read_parquet", path=dataset_url, storage_options={"anon": True})
+            eval_io(
+                "read_parquet",
+                path=dataset_url,
+                storage_options={"anon": True},
+                engine=engine,
+            )
 
+    @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",
         reason="The reason of tests fail in `cloud` mode is unknown for now - issue #3264",
     )
-    def test_read_parquet_without_metadata(self):
+    def test_read_parquet_without_metadata(self, engine):
         """Test that Modin can read parquet files not written by pandas."""
         from pyarrow import csv
         from pyarrow import parquet
@@ -1563,19 +1597,23 @@ def test_read_parquet_without_metadata(self):
             t = csv.read_csv(csv_fname)
             parquet.write_table(t, parquet_fname)
 
-            df_equals(
-                pd.read_parquet(parquet_fname), pandas.read_parquet(parquet_fname)
+            eval_io(
+                "read_parquet",
+                # read_parquet kwargs
+                path=parquet_fname,
+                engine=engine,
             )
         finally:
             teardown_test_files([parquet_fname, csv_fname])
 
-    def test_read_empty_parquet_file(self):
+    @pytest.mark.parametrize("engine", ["pyarrow", "fastparquet"])
+    def test_read_empty_parquet_file(self, engine):
         test_df = pandas.DataFrame()
         with tempfile.TemporaryDirectory() as directory:
             path = f"{directory}/data"
             os.makedirs(path)
             test_df.to_parquet(path + "/part-00000.parquet")
-            eval_io(fn_name="read_parquet", path=path)
+            eval_io(fn_name="read_parquet", path=path, engine=engine)
 
     @pytest.mark.xfail(
         condition="config.getoption('--simulate-cloud').lower() != 'off'",