Repository: modin-project/modin
Pull request: FEAT-#6767: Provide the ability to use experimental functionality when experimental mode is not enabled globally via an environment variable
<!--
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?

<!-- Please give a short brief about these changes. -->

Main changes:
* ~New `enable_exp_mode` context manager~
* New attribute for accessing experimental functionality: `modin = CachedAccessor("modin", ExperimentalFunctions)`
  - This is done in order to avoid adding new methods to the dataframe when importing module `modin.experimental.pandas`. ~When switching between two modes (Experimental and non experimental), it would be quite difficult to remove/add them again.~
  - It is not advisable to add experimental functions to the main namespace, so as not to mix them with pandas functions and confuse the users.
* ~As a side effect, the factory is now initialized more than once when using `enable_exp_mode` context manager.~

- [x] first commit message and PR title follow format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
  > **_NOTE:_**  If you edit the PR title to match this format, you need to add another commit (even if it's empty) or amend your last commit for the CI job that checks the PR title to pick up the new PR title.
- [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 #6767 <!-- issue must be created for each patch -->
- [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 -->


Problem statement:
Provide the ability to use experimental functionality when experimental mode is not enabled globally via an environment variable.
Example where it can be useful:

```python
import modin.pandas as pd

df = pd.DataFrame([1,2,3,4])
# [some code]

with modin.utils.enable_exp_mode():
    # this import has side effects that will need to be removed when leaving the context
    # for example:
    # 1. `IsExperimental.put(True)` 
    # 2. `setattr(DataFrame, "to_pickle_distributed", to_pickle_distributed)`
    # 3. Modification of internal factory and IO classes
    from modin.experimental.pandas import read_pickle_distributed, to_pickle_distributed
    to_pickle_distributed(df, "test_file*.pkl")
    # [some code]
    new_df = read_pickle_distributed("test_file*.pkl")
```

Hints from issue discussion:

Commit message: remove old implementation

Signed-off-by: Anatoly Myachev <anatoly.myachev@intel.com>

Patch to review:
diff --git a/docs/flow/modin/experimental/pandas.rst b/docs/flow/modin/experimental/pandas.rst
index 7036a9c5c24..d402e177cf8 100644
--- a/docs/flow/modin/experimental/pandas.rst
+++ b/docs/flow/modin/experimental/pandas.rst
@@ -13,4 +13,4 @@ Experimental API Reference
 .. autofunction:: read_csv_glob
 .. autofunction:: read_custom_text
 .. autofunction:: read_pickle_distributed
-.. automethod:: modin.experimental.pandas.DataFrame.to_pickle_distributed
+.. automethod:: modin.experimental.pandas.DataFrame._exp.to_pickle_distributed
diff --git a/docs/supported_apis/dataframe_supported.rst b/docs/supported_apis/dataframe_supported.rst
index 67cb3f9b603..227e042a0f1 100644
--- a/docs/supported_apis/dataframe_supported.rst
+++ b/docs/supported_apis/dataframe_supported.rst
@@ -424,7 +424,8 @@ default to pandas.
 +----------------------------+---------------------------+------------------------+----------------------------------------------------+
 | ``to_period``              | `to_period`_              | D                      |                                                    |
 +----------------------------+---------------------------+------------------------+----------------------------------------------------+
-| ``to_pickle``              | `to_pickle`_              | D                      | Experimental implementation: to_pickle_distributed |
+| ``to_pickle``              | `to_pickle`_              | D                      | Experimental implementation:                       |
+|                            |                           |                        | DataFrame._exp.to_pickle_distributed               |
 +----------------------------+---------------------------+------------------------+----------------------------------------------------+
 | ``to_records``             | `to_records`_             | D                      |                                                    |
 +----------------------------+---------------------------+------------------------+----------------------------------------------------+
diff --git a/docs/usage_guide/advanced_usage/index.rst b/docs/usage_guide/advanced_usage/index.rst
index 305f1a3da58..079ce055f66 100644
--- a/docs/usage_guide/advanced_usage/index.rst
+++ b/docs/usage_guide/advanced_usage/index.rst
@@ -31,7 +31,7 @@ Modin also supports these experimental APIs on top of pandas that are under acti
 - :py:func:`~modin.experimental.pandas.read_sql` -- add optional parameters for the database connection
 - :py:func:`~modin.experimental.pandas.read_custom_text` -- read custom text data from file
 - :py:func:`~modin.experimental.pandas.read_pickle_distributed`  -- read multiple files in a directory
-- :py:meth:`~modin.experimental.pandas.DataFrame.to_pickle_distributed` -- write to multiple files in a directory
+- :py:meth:`~modin.experimental.pandas.DataFrame._exp.to_pickle_distributed` -- write to multiple files in a directory
 
 DataFrame partitioning API
 --------------------------
diff --git a/modin/core/execution/dask/implementations/pandas_on_dask/io/io.py b/modin/core/execution/dask/implementations/pandas_on_dask/io/io.py
index 23cf38d9301..dcbf12744de 100644
--- a/modin/core/execution/dask/implementations/pandas_on_dask/io/io.py
+++ b/modin/core/execution/dask/implementations/pandas_on_dask/io/io.py
@@ -31,6 +31,9 @@
     SQLDispatcher,
 )
 from modin.core.storage_formats.pandas.parsers import (
+    ExperimentalCustomTextParser,
+    ExperimentalPandasPickleParser,
+    PandasCSVGlobParser,
     PandasCSVParser,
     PandasExcelParser,
     PandasFeatherParser,
@@ -40,6 +43,11 @@
     PandasSQLParser,
 )
 from modin.core.storage_formats.pandas.query_compiler import PandasQueryCompiler
+from modin.experimental.core.io import (
+    ExperimentalCSVGlobDispatcher,
+    ExperimentalCustomTextDispatcher,
+    ExperimentalPickleDispatcher,
+)
 
 
 class PandasOnDaskIO(BaseIO):
@@ -74,5 +82,15 @@ def __make_write(*classes, build_args=build_args):
     to_sql = __make_write(SQLDispatcher)
     read_excel = __make_read(PandasExcelParser, ExcelDispatcher)
 
+    # experimental methods that don't exist in pandas
+    read_csv_glob = __make_read(PandasCSVGlobParser, ExperimentalCSVGlobDispatcher)
+    read_pickle_distributed = __make_read(
+        ExperimentalPandasPickleParser, ExperimentalPickleDispatcher
+    )
+    to_pickle_distributed = __make_write(ExperimentalPickleDispatcher)
+    read_custom_text = __make_read(
+        ExperimentalCustomTextParser, ExperimentalCustomTextDispatcher
+    )
+
     del __make_read  # to not pollute class namespace
     del __make_write  # to not pollute class namespace
diff --git a/modin/core/execution/dispatching/factories/dispatcher.py b/modin/core/execution/dispatching/factories/dispatcher.py
index 623bf354d45..098f40b29d2 100644
--- a/modin/core/execution/dispatching/factories/dispatcher.py
+++ b/modin/core/execution/dispatching/factories/dispatcher.py
@@ -118,14 +118,13 @@ def get_factory(cls) -> factories.BaseFactory:
         return cls.__factory
 
     @classmethod
-    # FIXME: replace `_` parameter with `*args`
-    def _update_factory(cls, _):
+    def _update_factory(cls, *args):
         """
         Update and prepare factory with a new one specified via Modin config.
 
         Parameters
         ----------
-        _ : object
+        *args : iterable
             This parameters serves the compatibility purpose.
             Does not affect the result.
         """
@@ -200,14 +199,12 @@ def read_csv(cls, **kwargs):
         return cls.get_factory()._read_csv(**kwargs)
 
     @classmethod
-    @_inherit_docstrings(factories.ExperimentalPandasOnRayFactory._read_csv_glob)
+    @_inherit_docstrings(factories.PandasOnRayFactory._read_csv_glob)
     def read_csv_glob(cls, **kwargs):
         return cls.get_factory()._read_csv_glob(**kwargs)
 
     @classmethod
-    @_inherit_docstrings(
-        factories.ExperimentalPandasOnRayFactory._read_pickle_distributed
-    )
+    @_inherit_docstrings(factories.PandasOnRayFactory._read_pickle_distributed)
     def read_pickle_distributed(cls, **kwargs):
         return cls.get_factory()._read_pickle_distributed(**kwargs)
 
@@ -297,14 +294,12 @@ def to_pickle(cls, *args, **kwargs):
         return cls.get_factory()._to_pickle(*args, **kwargs)
 
     @classmethod
-    @_inherit_docstrings(
-        factories.ExperimentalPandasOnRayFactory._to_pickle_distributed
-    )
+    @_inherit_docstrings(factories.PandasOnRayFactory._to_pickle_distributed)
     def to_pickle_distributed(cls, *args, **kwargs):
         return cls.get_factory()._to_pickle_distributed(*args, **kwargs)
 
     @classmethod
-    @_inherit_docstrings(factories.ExperimentalPandasOnRayFactory._read_custom_text)
+    @_inherit_docstrings(factories.PandasOnRayFactory._read_custom_text)
     def read_custom_text(cls, **kwargs):
         return cls.get_factory()._read_custom_text(**kwargs)
 
diff --git a/modin/core/execution/dispatching/factories/factories.py b/modin/core/execution/dispatching/factories/factories.py
index 56f6d2c6dbf..29c99a2c2cd 100644
--- a/modin/core/execution/dispatching/factories/factories.py
+++ b/modin/core/execution/dispatching/factories/factories.py
@@ -26,9 +26,9 @@
 import pandas
 from pandas.util._decorators import doc
 
-from modin.config import Engine
+from modin.config import Engine, IsExperimental
 from modin.core.io import BaseIO
-from modin.utils import _inherit_docstrings, get_current_execution
+from modin.utils import get_current_execution
 
 _doc_abstract_factory_class = """
 Abstract {role} factory which allows to override the IO module easily.
@@ -329,6 +329,33 @@ def _read_pickle(cls, **kwargs):
         method="read_sql",
     )
     def _read_sql(cls, **kwargs):
+        if IsExperimental.get():
+            supported_engines = ("Ray", "Unidist", "Dask")
+            if Engine.get() not in supported_engines:
+                if "partition_column" in kwargs:
+                    if kwargs["partition_column"] is not None:
+                        warnings.warn(
+                            f"Distributed read_sql() was only implemented for {', '.join(supported_engines)} engines."
+                        )
+                    del kwargs["partition_column"]
+                if "lower_bound" in kwargs:
+                    if kwargs["lower_bound"] is not None:
+                        warnings.warn(
+                            f"Distributed read_sql() was only implemented for {', '.join(supported_engines)} engines."
+                        )
+                    del kwargs["lower_bound"]
+                if "upper_bound" in kwargs:
+                    if kwargs["upper_bound"] is not None:
+                        warnings.warn(
+                            f"Distributed read_sql() was only implemented for {', '.join(supported_engines)} engines."
+                        )
+                    del kwargs["upper_bound"]
+                if "max_sessions" in kwargs:
+                    if kwargs["max_sessions"] is not None:
+                        warnings.warn(
+                            f"Distributed read_sql() was only implemented for {', '.join(supported_engines)} engines."
+                        )
+                del kwargs["max_sessions"]
         return cls.io_cls.read_sql(**kwargs)
 
     @classmethod
@@ -427,86 +454,7 @@ def _to_parquet(cls, *args, **kwargs):
         """
         return cls.io_cls.to_parquet(*args, **kwargs)
 
-
-@doc(_doc_factory_class, execution_name="cuDFOnRay")
-class CudfOnRayFactory(BaseFactory):
-    @classmethod
-    @doc(_doc_factory_prepare_method, io_module_name="``cuDFOnRayIO``")
-    def prepare(cls):
-        from modin.core.execution.ray.implementations.cudf_on_ray.io import cuDFOnRayIO
-
-        cls.io_cls = cuDFOnRayIO
-
-
-@doc(_doc_factory_class, execution_name="PandasOnRay")
-class PandasOnRayFactory(BaseFactory):
-    @classmethod
-    @doc(_doc_factory_prepare_method, io_module_name="``PandasOnRayIO``")
-    def prepare(cls):
-        from modin.core.execution.ray.implementations.pandas_on_ray.io import (
-            PandasOnRayIO,
-        )
-
-        cls.io_cls = PandasOnRayIO
-
-
-@doc(_doc_factory_class, execution_name="PandasOnPython")
-class PandasOnPythonFactory(BaseFactory):
-    @classmethod
-    @doc(_doc_factory_prepare_method, io_module_name="``PandasOnPythonIO``")
-    def prepare(cls):
-        from modin.core.execution.python.implementations.pandas_on_python.io import (
-            PandasOnPythonIO,
-        )
-
-        cls.io_cls = PandasOnPythonIO
-
-
-@doc(_doc_factory_class, execution_name="PandasOnDask")
-class PandasOnDaskFactory(BaseFactory):
-    @classmethod
-    @doc(_doc_factory_prepare_method, io_module_name="``PandasOnDaskIO``")
-    def prepare(cls):
-        from modin.core.execution.dask.implementations.pandas_on_dask.io import (
-            PandasOnDaskIO,
-        )
-
-        cls.io_cls = PandasOnDaskIO
-
-
-@doc(_doc_abstract_factory_class, role="experimental")
-class ExperimentalBaseFactory(BaseFactory):
-    @classmethod
-    @_inherit_docstrings(BaseFactory._read_sql)
-    def _read_sql(cls, **kwargs):
-        supported_engines = ("Ray", "Unidist", "Dask")
-        if Engine.get() not in supported_engines:
-            if "partition_column" in kwargs:
-                if kwargs["partition_column"] is not None:
-                    warnings.warn(
-                        f"Distributed read_sql() was only implemented for {', '.join(supported_engines)} engines."
-                    )
-                del kwargs["partition_column"]
-            if "lower_bound" in kwargs:
-                if kwargs["lower_bound"] is not None:
-                    warnings.warn(
-                        f"Distributed read_sql() was only implemented for {', '.join(supported_engines)} engines."
-                    )
-                del kwargs["lower_bound"]
-            if "upper_bound" in kwargs:
-                if kwargs["upper_bound"] is not None:
-                    warnings.warn(
-                        f"Distributed read_sql() was only implemented for {', '.join(supported_engines)} engines."
-                    )
-                del kwargs["upper_bound"]
-            if "max_sessions" in kwargs:
-                if kwargs["max_sessions"] is not None:
-                    warnings.warn(
-                        f"Distributed read_sql() was only implemented for {', '.join(supported_engines)} engines."
-                    )
-                del kwargs["max_sessions"]
-        return cls.io_cls.read_sql(**kwargs)
-
+    # experimental methods that don't exist in pandas
     @classmethod
     @doc(
         _doc_io_method_raw_template,
@@ -569,37 +517,54 @@ def _to_pickle_distributed(cls, *args, **kwargs):
         return cls.io_cls.to_pickle_distributed(*args, **kwargs)
 
 
-@doc(_doc_factory_class, execution_name="experimental PandasOnRay")
-class ExperimentalPandasOnRayFactory(ExperimentalBaseFactory, PandasOnRayFactory):
+@doc(_doc_factory_class, execution_name="cuDFOnRay")
+class CudfOnRayFactory(BaseFactory):
+    @classmethod
+    @doc(_doc_factory_prepare_method, io_module_name="``cuDFOnRayIO``")
+    def prepare(cls):
+        from modin.core.execution.ray.implementations.cudf_on_ray.io import cuDFOnRayIO
+
+        cls.io_cls = cuDFOnRayIO
+
+
+@doc(_doc_factory_class, execution_name="PandasOnRay")
+class PandasOnRayFactory(BaseFactory):
     @classmethod
-    @doc(_doc_factory_prepare_method, io_module_name="``ExperimentalPandasOnRayIO``")
+    @doc(_doc_factory_prepare_method, io_module_name="``PandasOnRayIO``")
     def prepare(cls):
-        from modin.experimental.core.execution.ray.implementations.pandas_on_ray.io import (
-            ExperimentalPandasOnRayIO,
+        from modin.core.execution.ray.implementations.pandas_on_ray.io import (
+            PandasOnRayIO,
         )
 
-        cls.io_cls = ExperimentalPandasOnRayIO
+        cls.io_cls = PandasOnRayIO
 
 
-@doc(_doc_factory_class, execution_name="experimental PandasOnDask")
-class ExperimentalPandasOnDaskFactory(ExperimentalBaseFactory, PandasOnDaskFactory):
+@doc(_doc_factory_class, execution_name="PandasOnPython")
+class PandasOnPythonFactory(BaseFactory):
     @classmethod
-    @doc(_doc_factory_prepare_method, io_module_name="``ExperimentalPandasOnDaskIO``")
+    @doc(_doc_factory_prepare_method, io_module_name="``PandasOnPythonIO``")
     def prepare(cls):
-        from modin.experimental.core.execution.dask.implementations.pandas_on_dask.io import (
-            ExperimentalPandasOnDaskIO,
+        from modin.core.execution.python.implementations.pandas_on_python.io import (
+            PandasOnPythonIO,
         )
 
-        cls.io_cls = ExperimentalPandasOnDaskIO
+        cls.io_cls = PandasOnPythonIO
 
 
-@doc(_doc_factory_class, execution_name="experimental PandasOnPython")
-class ExperimentalPandasOnPythonFactory(ExperimentalBaseFactory, PandasOnPythonFactory):
-    pass
+@doc(_doc_factory_class, execution_name="PandasOnDask")
+class PandasOnDaskFactory(BaseFactory):
+    @classmethod
+    @doc(_doc_factory_prepare_method, io_module_name="``PandasOnDaskIO``")
+    def prepare(cls):
+        from modin.core.execution.dask.implementations.pandas_on_dask.io import (
+            PandasOnDaskIO,
+        )
+
+        cls.io_cls = PandasOnDaskIO
 
 
 @doc(_doc_factory_class, execution_name="experimental PyarrowOnRay")
-class ExperimentalPyarrowOnRayFactory(BaseFactory):  # pragma: no cover
+class PyarrowOnRayFactory(BaseFactory):  # pragma: no cover
     @classmethod
     @doc(_doc_factory_prepare_method, io_module_name="experimental ``PyarrowOnRayIO``")
     def prepare(cls):
@@ -611,7 +576,7 @@ def prepare(cls):
 
 
 @doc(_doc_factory_class, execution_name="experimental HdkOnNative")
-class ExperimentalHdkOnNativeFactory(BaseFactory):
+class HdkOnNativeFactory(BaseFactory):
     @classmethod
     @doc(_doc_factory_prepare_method, io_module_name="experimental ``HdkOnNativeIO``")
     def prepare(cls):
@@ -632,19 +597,3 @@ def prepare(cls):
         )
 
         cls.io_cls = PandasOnUnidistIO
-
-
-@doc(_doc_factory_class, execution_name="experimental PandasOnUnidist")
-class ExperimentalPandasOnUnidistFactory(
-    ExperimentalBaseFactory, PandasOnUnidistFactory
-):
-    @classmethod
-    @doc(
-        _doc_factory_prepare_method, io_module_name="``ExperimentalPandasOnUnidistIO``"
-    )
-    def prepare(cls):
-        from modin.experimental.core.execution.unidist.implementations.pandas_on_unidist.io import (
-            ExperimentalPandasOnUnidistIO,
-        )
-
-        cls.io_cls = ExperimentalPandasOnUnidistIO
diff --git a/modin/core/execution/ray/implementations/pandas_on_ray/io/io.py b/modin/core/execution/ray/implementations/pandas_on_ray/io/io.py
index c6409bd3e17..63e2578cdd2 100644
--- a/modin/core/execution/ray/implementations/pandas_on_ray/io/io.py
+++ b/modin/core/execution/ray/implementations/pandas_on_ray/io/io.py
@@ -30,6 +30,9 @@
     SQLDispatcher,
 )
 from modin.core.storage_formats.pandas.parsers import (
+    ExperimentalCustomTextParser,
+    ExperimentalPandasPickleParser,
+    PandasCSVGlobParser,
     PandasCSVParser,
     PandasExcelParser,
     PandasFeatherParser,
@@ -39,6 +42,11 @@
     PandasSQLParser,
 )
 from modin.core.storage_formats.pandas.query_compiler import PandasQueryCompiler
+from modin.experimental.core.io import (
+    ExperimentalCSVGlobDispatcher,
+    ExperimentalCustomTextDispatcher,
+    ExperimentalPickleDispatcher,
+)
 
 from ..dataframe import PandasOnRayDataframe
 from ..partitioning import PandasOnRayDataframePartition
@@ -76,6 +84,16 @@ def __make_write(*classes, build_args=build_args):
     to_sql = __make_write(SQLDispatcher)
     read_excel = __make_read(PandasExcelParser, ExcelDispatcher)
 
+    # experimental methods that don't exist in pandas
+    read_csv_glob = __make_read(PandasCSVGlobParser, ExperimentalCSVGlobDispatcher)
+    read_pickle_distributed = __make_read(
+        ExperimentalPandasPickleParser, ExperimentalPickleDispatcher
+    )
+    to_pickle_distributed = __make_write(ExperimentalPickleDispatcher)
+    read_custom_text = __make_read(
+        ExperimentalCustomTextParser, ExperimentalCustomTextDispatcher
+    )
+
     del __make_read  # to not pollute class namespace
     del __make_write  # to not pollute class namespace
 
diff --git a/modin/core/execution/unidist/implementations/pandas_on_unidist/io/io.py b/modin/core/execution/unidist/implementations/pandas_on_unidist/io/io.py
index df90e6ac068..cf540fabc84 100644
--- a/modin/core/execution/unidist/implementations/pandas_on_unidist/io/io.py
+++ b/modin/core/execution/unidist/implementations/pandas_on_unidist/io/io.py
@@ -29,6 +29,9 @@
     SQLDispatcher,
 )
 from modin.core.storage_formats.pandas.parsers import (
+    ExperimentalCustomTextParser,
+    ExperimentalPandasPickleParser,
+    PandasCSVGlobParser,
     PandasCSVParser,
     PandasExcelParser,
     PandasFeatherParser,
@@ -38,6 +41,11 @@
     PandasSQLParser,
 )
 from modin.core.storage_formats.pandas.query_compiler import PandasQueryCompiler
+from modin.experimental.core.io import (
+    ExperimentalCSVGlobDispatcher,
+    ExperimentalCustomTextDispatcher,
+    ExperimentalPickleDispatcher,
+)
 
 from ..dataframe import PandasOnUnidistDataframe
 from ..partitioning import PandasOnUnidistDataframePartition
@@ -75,6 +83,16 @@ def __make_write(*classes, build_args=build_args):
     to_sql = __make_write(SQLDispatcher)
     read_excel = __make_read(PandasExcelParser, ExcelDispatcher)
 
+    # experimental methods that don't exist in pandas
+    read_csv_glob = __make_read(PandasCSVGlobParser, ExperimentalCSVGlobDispatcher)
+    read_pickle_distributed = __make_read(
+        ExperimentalPandasPickleParser, ExperimentalPickleDispatcher
+    )
+    to_pickle_distributed = __make_write(ExperimentalPickleDispatcher)
+    read_custom_text = __make_read(
+        ExperimentalCustomTextParser, ExperimentalCustomTextDispatcher
+    )
+
     del __make_read  # to not pollute class namespace
     del __make_write  # to not pollute class namespace
 
diff --git a/modin/core/io/sql/sql_dispatcher.py b/modin/core/io/sql/sql_dispatcher.py
index 782207e95d7..8225a0bb887 100644
--- a/modin/core/io/sql/sql_dispatcher.py
+++ b/modin/core/io/sql/sql_dispatcher.py
@@ -20,11 +20,13 @@
 """
 
 import math
+import warnings
+from typing import Any
 
 import numpy as np
 import pandas
 
-from modin.config import NPartitions, ReadSqlEngine
+from modin.config import IsExperimental, NPartitions, ReadSqlEngine
 from modin.core.io.file_dispatcher import FileDispatcher
 from modin.db_conn import ModinDatabaseConnection
 
@@ -176,3 +178,118 @@ def func(df):  # pragma: no cover
         cls.materialize(
             [part.list_of_blocks[0] for row in result._partitions for part in row]
         )
+
+    def __getattribute__(self, __name: str) -> Any:
+        if __name == "_read" and IsExperimental.get():
+            __name = "_read_exp"
+        return super().__getattribute__(__name)
+
+    # experimental implementation
+    __read_sql_with_offset = None
+
+    @classmethod
+    def preprocess_func(cls):  # noqa: RT01
+        """Prepare a function for transmission to remote workers."""
+        if cls.__read_sql_with_offset is None:
+            # sql deps are optional, so import only when needed
+            from modin.core.io.sql.utils import read_sql_with_offset
+
+            cls.__read_sql_with_offset = cls.put(read_sql_with_offset)
+        return cls.__read_sql_with_offset
+
+    @classmethod
+    def _read_exp(
+        cls,
+        sql,
+        con,
+        index_col,
+        coerce_float,
+        params,
+        parse_dates,
+        columns,
+        chunksize,
+        dtype_backend,
+        dtype,
+        partition_column,
+        lower_bound,
+        upper_bound,
+        max_sessions,
+    ):  # noqa: PR01
+        """
+        Read SQL query or database table into a DataFrame.
+
+        Documentation for parameters can be found at `modin.read_sql`.
+
+        Returns
+        -------
+        BaseQueryCompiler
+            A new query compiler with imported data for further processing.
+        """
+        # sql deps are optional, so import only when needed
+        from modin.core.io.sql.utils import get_query_info, is_distributed
+
+        if not is_distributed(partition_column, lower_bound, upper_bound):
+            message = "Defaulting to Modin core implementation; \
+                'partition_column', 'lower_bound', 'upper_bound' must be different from None"
+            warnings.warn(message)
+            return cls.base_io.read_sql(
+                sql,
+                con,
+                index_col,
+                coerce_float=coerce_float,
+                params=params,
+                parse_dates=parse_dates,
+                columns=columns,
+                chunksize=chunksize,
+                dtype_backend=dtype_backend,
+                dtype=dtype,
+            )
+        #  starts the distributed alternative
+        cols_names, query = get_query_info(sql, con, partition_column)
+        num_parts = min(NPartitions.get(), max_sessions if max_sessions else 1)
+        num_splits = min(len(cols_names), num_parts)
+        diff = (upper_bound - lower_bound) + 1
+        min_size = diff // num_parts
+        rest = diff % num_parts
+        partition_ids = []
+        index_ids = []
+        end = lower_bound - 1
+        func = cls.preprocess_func()
+        for part in range(num_parts):
+            if rest:
+                size = min_size + 1
+                rest -= 1
+            else:
+                size = min_size
+            start = end + 1
+            end = start + size - 1
+            partition_id = cls.deploy(
+                func,
+                f_args=(
+                    partition_column,
+                    start,
+                    end,
+                    num_splits,
+                    query,
+                    con,
+                    index_col,
+                    coerce_float,
+                    params,
+                    parse_dates,
+                    columns,
+                    chunksize,
+                    dtype_backend,
+                    dtype,
+                ),
+                num_returns=num_splits + 1,
+            )
+            partition_ids.append(
+                [cls.frame_partition_cls(obj) for obj in partition_id[:-1]]
+            )
+            index_ids.append(partition_id[-1])
+        new_index = pandas.RangeIndex(sum(cls.materialize(index_ids)))
+        new_query_compiler = cls.query_compiler_cls(
+            cls.frame_cls(np.array(partition_ids), new_index, cols_names)
+        )
+        new_query_compiler._modin_frame.synchronize_labels(axis=0)
+        return new_query_compiler
diff --git a/modin/experimental/core/io/sql/utils.py b/modin/core/io/sql/utils.py
similarity index 100%
rename from modin/experimental/core/io/sql/utils.py
rename to modin/core/io/sql/utils.py
diff --git a/modin/experimental/core/execution/dask/implementations/pandas_on_dask/io/__init__.py b/modin/experimental/core/execution/dask/implementations/pandas_on_dask/io/__init__.py
deleted file mode 100644
index ae42d666b91..00000000000
--- a/modin/experimental/core/execution/dask/implementations/pandas_on_dask/io/__init__.py
+++ /dev/null
@@ -1,18 +0,0 @@
-# Licensed to Modin Development Team under one or more contributor license agreements.
-# See the NOTICE file distributed with this work for additional information regarding
-# copyright ownership.  The Modin Development Team licenses this file to you under the
-# Apache License, Version 2.0 (the "License"); you may not use this file except in
-# compliance with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software distributed under
-# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
-# ANY KIND, either express or implied. See the License for the specific language
-# governing permissions and limitations under the License.
-
-"""Experimental Base IO classes optimized for pandas on Dask execution."""
-
-from .io import ExperimentalPandasOnDaskIO
-
-__all__ = ["ExperimentalPandasOnDaskIO"]
diff --git a/modin/experimental/core/execution/dask/implementations/pandas_on_dask/io/io.py b/modin/experimental/core/execution/dask/implementations/pandas_on_dask/io/io.py
deleted file mode 100644
index bfc628eebb8..00000000000
--- a/modin/experimental/core/execution/dask/implementations/pandas_on_dask/io/io.py
+++ /dev/null
@@ -1,77 +0,0 @@
-# Licensed to Modin Development Team under one or more contributor license agreements.
-# See the NOTICE file distributed with this work for additional information regarding
-# copyright ownership.  The Modin Development Team licenses this file to you under the
-# Apache License, Version 2.0 (the "License"); you may not use this file except in
-# compliance with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software distributed under
-# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
-# ANY KIND, either express or implied. See the License for the specific language
-# governing permissions and limitations under the License.
-
-"""
-Module houses experimental IO classes and parser functions needed for these classes.
-
-Any function or class can be considered experimental API if it is not strictly replicating existent
-Query Compiler API, even if it is only extending the API.
-"""
-
-from modin.core.execution.dask.common import DaskWrapper
-from modin.core.execution.dask.implementations.pandas_on_dask.dataframe import (
-    PandasOnDaskDataframe,
-)
-from modin.core.execution.dask.implementations.pandas_on_dask.io import PandasOnDaskIO
-from modin.core.execution.dask.implementations.pandas_on_dask.partitioning import (
-    PandasOnDaskDataframePartition,
-)
-from modin.core.storage_formats.pandas.parsers import (
-    ExperimentalCustomTextParser,
-    ExperimentalPandasPickleParser,
-    PandasCSVGlobParser,
-)
-from modin.core.storage_formats.pandas.query_compiler import PandasQueryCompiler
-from modin.experimental.core.io import (
-    ExperimentalCSVGlobDispatcher,
-    ExperimentalCustomTextDispatcher,
-    ExperimentalPickleDispatcher,
-    ExperimentalSQLDispatcher,
-)
-
-
-class ExperimentalPandasOnDaskIO(PandasOnDaskIO):
-    """
-    Class for handling experimental IO functionality with pandas storage format and Dask engine.
-
-    ``ExperimentalPandasOnDaskIO`` inherits some util functions and unmodified IO functions
-    from ``PandasOnDaskIO`` class.
-    """
-
-    build_args = dict(
-        frame_partition_cls=PandasOnDaskDataframePartition,
-        query_compiler_cls=PandasQueryCompiler,
-        frame_cls=PandasOnDaskDataframe,
-        base_io=PandasOnDaskIO,
-    )
-
-    def __make_read(*classes, build_args=build_args):
-        # used to reduce code duplication
-        return type("", (DaskWrapper, *classes), build_args).read
-
-    def __make_write(*classes, build_args=build_args):
-        # used to reduce code duplication
-        return type("", (DaskWrapper, *classes), build_args).write
-
-    read_csv_glob = __make_read(PandasCSVGlobParser, ExperimentalCSVGlobDispatcher)
-    read_pickle_distributed = __make_read(
-        ExperimentalPandasPickleParser, ExperimentalPickleDispatcher
-    )
-    to_pickle_distributed = __make_write(ExperimentalPickleDispatcher)
-    read_custom_text = __make_read(
-        ExperimentalCustomTextParser, ExperimentalCustomTextDispatcher
-    )
-    read_sql = __make_read(ExperimentalSQLDispatcher)
-
-    del __make_read  # to not pollute class namespace
-    del __make_write  # to not pollute class namespace
diff --git a/modin/experimental/core/execution/ray/implementations/pandas_on_ray/__init__.py b/modin/experimental/core/execution/ray/implementations/pandas_on_ray/__init__.py
deleted file mode 100644
index bccf5e24948..00000000000
--- a/modin/experimental/core/execution/ray/implementations/pandas_on_ray/__init__.py
+++ /dev/null
@@ -1,14 +0,0 @@
-# Licensed to Modin Development Team under one or more contributor license agreements.
-# See the NOTICE file distributed with this work for additional information regarding
-# copyright ownership.  The Modin Development Team licenses this file to you under the
-# Apache License, Version 2.0 (the "License"); you may not use this file except in
-# compliance with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software distributed under
-# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
-# ANY KIND, either express or implied. See the License for the specific language
-# governing permissions and limitations under the License.
-
-"""Experimental functionality related to Ray execution engine and optimized for pandas storage format."""
diff --git a/modin/experimental/core/execution/ray/implementations/pandas_on_ray/io/__init__.py b/modin/experimental/core/execution/ray/implementations/pandas_on_ray/io/__init__.py
deleted file mode 100644
index 50871daeadd..00000000000
--- a/modin/experimental/core/execution/ray/implementations/pandas_on_ray/io/__init__.py
+++ /dev/null
@@ -1,18 +0,0 @@
-# Licensed to Modin Development Team under one or more contributor license agreements.
-# See the NOTICE file distributed with this work for additional information regarding
-# copyright ownership.  The Modin Development Team licenses this file to you under the
-# Apache License, Version 2.0 (the "License"); you may not use this file except in
-# compliance with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software distributed under
-# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
-# ANY KIND, either express or implied. See the License for the specific language
-# governing permissions and limitations under the License.
-
-"""Experimental Base IO classes optimized for pandas on Ray execution."""
-
-from .io import ExperimentalPandasOnRayIO
-
-__all__ = ["ExperimentalPandasOnRayIO"]
diff --git a/modin/experimental/core/execution/ray/implementations/pandas_on_ray/io/io.py b/modin/experimental/core/execution/ray/implementations/pandas_on_ray/io/io.py
deleted file mode 100644
index 31fbd6d92b3..00000000000
--- a/modin/experimental/core/execution/ray/implementations/pandas_on_ray/io/io.py
+++ /dev/null
@@ -1,77 +0,0 @@
-# Licensed to Modin Development Team under one or more contributor license agreements.
-# See the NOTICE file distributed with this work for additional information regarding
-# copyright ownership.  The Modin Development Team licenses this file to you under the
-# Apache License, Version 2.0 (the "License"); you may not use this file except in
-# compliance with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software distributed under
-# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
-# ANY KIND, either express or implied. See the License for the specific language
-# governing permissions and limitations under the License.
-
-"""
-Module houses experimental IO classes and parser functions needed for these classes.
-
-Any function or class can be considered experimental API if it is not strictly replicating existent
-Query Compiler API, even if it is only extending the API.
-"""
-
-from modin.core.execution.ray.common import RayWrapper
-from modin.core.execution.ray.implementations.pandas_on_ray.dataframe import (
-    PandasOnRayDataframe,
-)
-from modin.core.execution.ray.implementations.pandas_on_ray.io import PandasOnRayIO
-from modin.core.execution.ray.implementations.pandas_on_ray.partitioning import (
-    PandasOnRayDataframePartition,
-)
-from modin.core.storage_formats.pandas.parsers import (
-    ExperimentalCustomTextParser,
-    ExperimentalPandasPickleParser,
-    PandasCSVGlobParser,
-)
-from modin.core.storage_formats.pandas.query_compiler import PandasQueryCompiler
-from modin.experimental.core.io import (
-    ExperimentalCSVGlobDispatcher,
-    ExperimentalCustomTextDispatcher,
-    ExperimentalPickleDispatcher,
-    ExperimentalSQLDispatcher,
-)
-
-
-class ExperimentalPandasOnRayIO(PandasOnRayIO):
-    """
-    Class for handling experimental IO functionality with pandas storage format and Ray engine.
-
-    ``ExperimentalPandasOnRayIO`` inherits some util functions and unmodified IO functions
-    from ``PandasOnRayIO`` class.
-    """
-
-    build_args = dict(
-        frame_partition_cls=PandasOnRayDataframePartition,
-        query_compiler_cls=PandasQueryCompiler,
-        frame_cls=PandasOnRayDataframe,
-        base_io=PandasOnRayIO,
-    )
-
-    def __make_read(*classes, build_args=build_args):
-        # used to reduce code duplication
-        return type("", (RayWrapper, *classes), build_args).read
-
-    def __make_write(*classes, build_args=build_args):
-        # used to reduce code duplication
-        return type("", (RayWrapper, *classes), build_args).write
-
-    read_csv_glob = __make_read(PandasCSVGlobParser, ExperimentalCSVGlobDispatcher)
-    read_pickle_distributed = __make_read(
-        ExperimentalPandasPickleParser, ExperimentalPickleDispatcher
-    )
-    to_pickle_distributed = __make_write(ExperimentalPickleDispatcher)
-    read_custom_text = __make_read(
-        ExperimentalCustomTextParser, ExperimentalCustomTextDispatcher
-    )
-    read_sql = __make_read(ExperimentalSQLDispatcher)
-
-    del __make_read  # to not pollute class namespace
-    del __make_write  # to not pollute class namespace
diff --git a/modin/experimental/core/execution/unidist/implementations/pandas_on_unidist/io/__init__.py b/modin/experimental/core/execution/unidist/implementations/pandas_on_unidist/io/__init__.py
deleted file mode 100644
index fd611287f7c..00000000000
--- a/modin/experimental/core/execution/unidist/implementations/pandas_on_unidist/io/__init__.py
+++ /dev/null
@@ -1,18 +0,0 @@
-# Licensed to Modin Development Team under one or more contributor license agreements.
-# See the NOTICE file distributed with this work for additional information regarding
-# copyright ownership.  The Modin Development Team licenses this file to you under the
-# Apache License, Version 2.0 (the "License"); you may not use this file except in
-# compliance with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software distributed under
-# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
-# ANY KIND, either express or implied. See the License for the specific language
-# governing permissions and limitations under the License.
-
-"""Experimental Base IO classes optimized for pandas on unidist execution."""
-
-from .io import ExperimentalPandasOnUnidistIO
-
-__all__ = ["ExperimentalPandasOnUnidistIO"]
diff --git a/modin/experimental/core/execution/unidist/implementations/pandas_on_unidist/io/io.py b/modin/experimental/core/execution/unidist/implementations/pandas_on_unidist/io/io.py
deleted file mode 100644
index c94b0621a4d..00000000000
--- a/modin/experimental/core/execution/unidist/implementations/pandas_on_unidist/io/io.py
+++ /dev/null
@@ -1,79 +0,0 @@
-# Licensed to Modin Development Team under one or more contributor license agreements.
-# See the NOTICE file distributed with this work for additional information regarding
-# copyright ownership.  The Modin Development Team licenses this file to you under the
-# Apache License, Version 2.0 (the "License"); you may not use this file except in
-# compliance with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software distributed under
-# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
-# ANY KIND, either express or implied. See the License for the specific language
-# governing permissions and limitations under the License.
-
-"""
-Module houses experimental IO classes and parser functions needed for these classes.
-
-Any function or class can be considered experimental API if it is not strictly replicating existent
-Query Compiler API, even if it is only extending the API.
-"""
-
-from modin.core.execution.unidist.common import UnidistWrapper
-from modin.core.execution.unidist.implementations.pandas_on_unidist.dataframe import (
-    PandasOnUnidistDataframe,
-)
-from modin.core.execution.unidist.implementations.pandas_on_unidist.io import (
-    PandasOnUnidistIO,
-)
-from modin.core.execution.unidist.implementations.pandas_on_unidist.partitioning import (
-    PandasOnUnidistDataframePartition,
-)
-from modin.core.storage_formats.pandas.parsers import (
-    ExperimentalCustomTextParser,
-    ExperimentalPandasPickleParser,
-    PandasCSVGlobParser,
-)
-from modin.core.storage_formats.pandas.query_compiler import PandasQueryCompiler
-from modin.experimental.core.io import (
-    ExperimentalCSVGlobDispatcher,
-    ExperimentalCustomTextDispatcher,
-    ExperimentalPickleDispatcher,
-    ExperimentalSQLDispatcher,
-)
-
-
-class ExperimentalPandasOnUnidistIO(PandasOnUnidistIO):
-    """
-    Class for handling experimental IO functionality with pandas storage format and unidist engine.
-
-    ``ExperimentalPandasOnUnidistIO`` inherits some util functions and unmodified IO functions
-    from ``PandasOnUnidistIO`` class.
-    """
-
-    build_args = dict(
-        frame_partition_cls=PandasOnUnidistDataframePartition,
-        query_compiler_cls=PandasQueryCompiler,
-        frame_cls=PandasOnUnidistDataframe,
-        base_io=PandasOnUnidistIO,
-    )
-
-    def __make_read(*classes, build_args=build_args):
-        # used to reduce code duplication
-        return type("", (UnidistWrapper, *classes), build_args).read
-
-    def __make_write(*classes, build_args=build_args):
-        # used to reduce code duplication
-        return type("", (UnidistWrapper, *classes), build_args).write
-
-    read_csv_glob = __make_read(PandasCSVGlobParser, ExperimentalCSVGlobDispatcher)
-    read_pickle_distributed = __make_read(
-        ExperimentalPandasPickleParser, ExperimentalPickleDispatcher
-    )
-    to_pickle_distributed = __make_write(ExperimentalPickleDispatcher)
-    read_custom_text = __make_read(
-        ExperimentalCustomTextParser, ExperimentalCustomTextDispatcher
-    )
-    read_sql = __make_read(ExperimentalSQLDispatcher)
-
-    del __make_read  # to not pollute class namespace
-    del __make_write  # to not pollute class namespace
diff --git a/modin/experimental/core/io/__init__.py b/modin/experimental/core/io/__init__.py
index 9f3c64b00a8..fe75f8c1421 100644
--- a/modin/experimental/core/io/__init__.py
+++ b/modin/experimental/core/io/__init__.py
@@ -14,13 +14,11 @@
 """Experimental IO functions implementations."""
 
 from .pickle.pickle_dispatcher import ExperimentalPickleDispatcher
-from .sql.sql_dispatcher import ExperimentalSQLDispatcher
 from .text.csv_glob_dispatcher import ExperimentalCSVGlobDispatcher
 from .text.custom_text_dispatcher import ExperimentalCustomTextDispatcher
 
 __all__ = [
     "ExperimentalCSVGlobDispatcher",
-    "ExperimentalSQLDispatcher",
     "ExperimentalPickleDispatcher",
     "ExperimentalCustomTextDispatcher",
 ]
diff --git a/modin/experimental/core/io/sql/__init__.py b/modin/experimental/core/io/sql/__init__.py
deleted file mode 100644
index b48459f4d43..00000000000
--- a/modin/experimental/core/io/sql/__init__.py
+++ /dev/null
@@ -1,14 +0,0 @@
-# Licensed to Modin Development Team under one or more contributor license agreements.
-# See the NOTICE file distributed with this work for additional information regarding
-# copyright ownership.  The Modin Development Team licenses this file to you under the
-# Apache License, Version 2.0 (the "License"); you may not use this file except in
-# compliance with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software distributed under
-# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
-# ANY KIND, either express or implied. See the License for the specific language
-# governing permissions and limitations under the License.
-
-"""Experimental SQL format type IO functions implementations."""
diff --git a/modin/experimental/core/io/sql/sql_dispatcher.py b/modin/experimental/core/io/sql/sql_dispatcher.py
deleted file mode 100644
index bc6285c4107..00000000000
--- a/modin/experimental/core/io/sql/sql_dispatcher.py
+++ /dev/null
@@ -1,135 +0,0 @@
-# Licensed to Modin Development Team under one or more contributor license agreements.
-# See the NOTICE file distributed with this work for additional information regarding
-# copyright ownership.  The Modin Development Team licenses this file to you under the
-# Apache License, Version 2.0 (the "License"); you may not use this file except in
-# compliance with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software distributed under
-# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
-# ANY KIND, either express or implied. See the License for the specific language
-# governing permissions and limitations under the License.
-
-"""Module houses `ExperimentalSQLDispatcher` class."""
-
-import warnings
-
-import numpy as np
-import pandas
-
-from modin.config import NPartitions
-from modin.core.io import SQLDispatcher
-
-
-class ExperimentalSQLDispatcher(SQLDispatcher):
-    """Class handles experimental utils for reading SQL queries or database tables."""
-
-    __read_sql_with_offset = None
-
-    @classmethod
-    def preprocess_func(cls):  # noqa: RT01
-        """Prepare a function for transmission to remote workers."""
-        if cls.__read_sql_with_offset is None:
-            # sql deps are optional, so import only when needed
-            from modin.experimental.core.io.sql.utils import read_sql_with_offset
-
-            cls.__read_sql_with_offset = cls.put(read_sql_with_offset)
-        return cls.__read_sql_with_offset
-
-    @classmethod
-    def _read(
-        cls,
-        sql,
-        con,
-        index_col,
-        coerce_float,
-        params,
-        parse_dates,
-        columns,
-        chunksize,
-        dtype_backend,
-        dtype,
-        partition_column,
-        lower_bound,
-        upper_bound,
-        max_sessions,
-    ):  # noqa: PR01
-        """
-        Read SQL query or database table into a DataFrame.
-
-        Documentation for parameters can be found at `modin.read_sql`.
-
-        Returns
-        -------
-        BaseQueryCompiler
-            A new query compiler with imported data for further processing.
-        """
-        # sql deps are optional, so import only when needed
-        from modin.experimental.core.io.sql.utils import get_query_info, is_distributed
-
-        if not is_distributed(partition_column, lower_bound, upper_bound):
-            message = "Defaulting to Modin core implementation; \
-                'partition_column', 'lower_bound', 'upper_bound' must be different from None"
-            warnings.warn(message)
-            return cls.base_io.read_sql(
-                sql,
-                con,
-                index_col,
-                coerce_float=coerce_float,
-                params=params,
-                parse_dates=parse_dates,
-                columns=columns,
-                chunksize=chunksize,
-                dtype_backend=dtype_backend,
-                dtype=dtype,
-            )
-        #  starts the distributed alternative
-        cols_names, query = get_query_info(sql, con, partition_column)
-        num_parts = min(NPartitions.get(), max_sessions if max_sessions else 1)
-        num_splits = min(len(cols_names), num_parts)
-        diff = (upper_bound - lower_bound) + 1
-        min_size = diff // num_parts
-        rest = diff % num_parts
-        partition_ids = []
-        index_ids = []
-        end = lower_bound - 1
-        func = cls.preprocess_func()
-        for part in range(num_parts):
-            if rest:
-                size = min_size + 1
-                rest -= 1
-            else:
-                size = min_size
-            start = end + 1
-            end = start + size - 1
-            partition_id = cls.deploy(
-                func,
-                f_args=(
-                    partition_column,
-                    start,
-                    end,
-                    num_splits,
-                    query,
-                    con,
-                    index_col,
-                    coerce_float,
-                    params,
-                    parse_dates,
-                    columns,
-                    chunksize,
-                    dtype_backend,
-                    dtype,
-                ),
-                num_returns=num_splits + 1,
-            )
-            partition_ids.append(
-                [cls.frame_partition_cls(obj) for obj in partition_id[:-1]]
-            )
-            index_ids.append(partition_id[-1])
-        new_index = pandas.RangeIndex(sum(cls.materialize(index_ids)))
-        new_query_compiler = cls.query_compiler_cls(
-            cls.frame_cls(np.array(partition_ids), new_index, cols_names)
-        )
-        new_query_compiler._modin_frame.synchronize_labels(axis=0)
-        return new_query_compiler
diff --git a/modin/experimental/pandas/__init__.py b/modin/experimental/pandas/__init__.py
index 0deba579ecb..a0b46970b9d 100644
--- a/modin/experimental/pandas/__init__.py
+++ b/modin/experimental/pandas/__init__.py
@@ -32,6 +32,8 @@
 >>> df = pd.read_csv_glob("data*.csv")
 """
 
+import functools
+
 from modin.config import IsExperimental
 
 IsExperimental.put(True)
@@ -48,6 +50,19 @@
     to_pickle_distributed,
 )
 
+old_to_pickle_distributed = to_pickle_distributed
+
+
+@functools.wraps(to_pickle_distributed)
+def to_pickle_distributed(*args, **kwargs):
+    warnings.warn(
+        "`DataFrame.to_pickle_distributed` will be removed after Modin 0.26.0 release. "
+        + "Please use `DataFrame._exp.to_pickle_distributed` instead.",
+        category=FutureWarning,
+    )
+    return old_to_pickle_distributed(*args, **kwargs)
+
+
 setattr(DataFrame, "to_pickle_distributed", to_pickle_distributed)  # noqa: F405
 
 warnings.warn(
diff --git a/modin/experimental/pandas/io.py b/modin/experimental/pandas/io.py
index fb3b1df1c42..b204b8a8c30 100644
--- a/modin/experimental/pandas/io.py
+++ b/modin/experimental/pandas/io.py
@@ -316,7 +316,7 @@ def read_pickle_distributed(
 
     This experimental feature provides parallel reading from multiple pickle files which are
     defined by glob pattern. The files must contain parts of one dataframe, which can be
-    obtained, for example, by `to_pickle_distributed` function.
+    obtained, for example, by `DataFrame._exp.to_pickle_distributed` function.
 
     Parameters
     ----------
diff --git a/modin/experimental/pandas/test/test_io_exp.py b/modin/experimental/pandas/test/test_io_exp.py
index 4bfd8c9b7ab..f9ac08bef91 100644
--- a/modin/experimental/pandas/test/test_io_exp.py
+++ b/modin/experimental/pandas/test/test_io_exp.py
@@ -260,7 +260,7 @@ def test_distributed_pickling(filename, compression):
         if filename_param == test_default_to_pickle_filename
         else contextlib.nullcontext()
     ):
-        df.to_pickle_distributed(filename, compression=compression)
+        df._exp.to_pickle_distributed(filename, compression=compression)
         pickled_df = pd.read_pickle_distributed(filename, compression=compression)
     df_equals(pickled_df, df)
 
diff --git a/modin/pandas/accessor.py b/modin/pandas/accessor.py
index 611175166cf..a78c9a3e1a9 100644
--- a/modin/pandas/accessor.py
+++ b/modin/pandas/accessor.py
@@ -21,7 +21,11 @@
 CachedAccessor implements API of pandas.core.accessor.CachedAccessor
 """
 
+import pickle
+import warnings
+
 import pandas
+from pandas._typing import CompressionOptions, StorageOptions
 from pandas.core.dtypes.dtypes import SparseDtype
 
 from modin import pandas as pd
@@ -191,3 +195,71 @@ def __get__(self, obj, cls):
         accessor_obj = self._accessor(obj)
         object.__setattr__(obj, self._name, accessor_obj)
         return accessor_obj
+
+
+class ExperimentalFunctions:
+    """
+    Namespace class for accessing experimental Modin functions.
+
+    Parameters
+    ----------
+    data : DataFrame or Series
+        Object to operate on.
+    """
+
+    def __init__(self, data):
+        warnings.warn(
+            "Thank you for using the Modin Experimental pandas API."
+            + "\nPlease note that some of these APIs deviate from pandas in order to "
+            + "provide improved performance."
+        )
+        self._data = data
+
+    def to_pickle_distributed(
+        self,
+        filepath_or_buffer,
+        compression: CompressionOptions = "infer",
+        protocol: int = pickle.HIGHEST_PROTOCOL,
+        storage_options: StorageOptions = None,
+    ):
+        """
+        Pickle (serialize) object to file.
+
+        This experimental feature provides parallel writing into multiple pickle files which are
+        defined by glob pattern, otherwise (without glob pattern) default pandas implementation is used.
+
+        Parameters
+        ----------
+        filepath_or_buffer : str, path object or file-like object
+            File path where the pickled object will be stored.
+        compression : {{'infer', 'gzip', 'bz2', 'zip', 'xz', None}}, default: 'infer'
+            A string representing the compression to use in the output file. By
+            default, infers from the file extension in specified path.
+            Compression mode may be any of the following possible
+            values: {{'infer', 'gzip', 'bz2', 'zip', 'xz', None}}. If compression
+            mode is 'infer' and path_or_buf is path-like, then detect
+            compression mode from the following extensions:
+            '.gz', '.bz2', '.zip' or '.xz'. (otherwise no compression).
+            If dict given and mode is 'zip' or inferred as 'zip', other entries
+            passed as additional compression options.
+        protocol : int, default: pickle.HIGHEST_PROTOCOL
+            Int which indicates which protocol should be used by the pickler,
+            default HIGHEST_PROTOCOL (see `pickle docs <https://docs.python.org/3/library/pickle.html>`_
+            paragraph 12.1.2 for details). The possible  values are 0, 1, 2, 3, 4, 5. A negative value
+            for the protocol parameter is equivalent to setting its value to HIGHEST_PROTOCOL.
+        storage_options : dict, optional
+            Extra options that make sense for a particular storage connection, e.g.
+            host, port, username, password, etc., if using a URL that will be parsed by
+            fsspec, e.g., starting "s3://", "gcs://". An error will be raised if providing
+            this argument with a non-fsspec URL. See the fsspec and backend storage
+            implementation docs for the set of allowed keys and values.
+        """
+        from modin.experimental.pandas.io import to_pickle_distributed
+
+        to_pickle_distributed(
+            self._data,
+            filepath_or_buffer=filepath_or_buffer,
+            compression=compression,
+            protocol=protocol,
+            storage_options=storage_options,
+        )
diff --git a/modin/pandas/dataframe.py b/modin/pandas/dataframe.py
index 2484b3932e2..2a3b82f4c36 100644
--- a/modin/pandas/dataframe.py
+++ b/modin/pandas/dataframe.py
@@ -53,7 +53,7 @@
     try_cast_to_pandas,
 )
 
-from .accessor import CachedAccessor, SparseFrameAccessor
+from .accessor import CachedAccessor, ExperimentalFunctions, SparseFrameAccessor
 from .base import _ATTRS_NO_LOOKUP, BasePandasDataset
 from .groupby import DataFrameGroupBy
 from .iterator import PartitionIterator
@@ -3169,3 +3169,6 @@ def __reduce__(self):
         return self._inflate_light, (self._query_compiler, pid)
 
     # Persistance support methods - END
+
+    # Namespace for experimental functions
+    _exp = CachedAccessor("_exp", ExperimentalFunctions)
diff --git a/modin/utils.py b/modin/utils.py
index c42a0516fa3..82533a397bc 100644
--- a/modin/utils.py
+++ b/modin/utils.py
@@ -14,6 +14,7 @@
 """Collection of general utility functions, mostly for internal use."""
 
 import codecs
+import contextlib
 import functools
 import importlib
 import inspect
@@ -27,6 +28,7 @@
 from typing import (
     Any,
     Callable,
+    Generator,
     Iterable,
     List,
     Mapping,
@@ -695,7 +697,7 @@ def get_current_execution() -> str:
     str
         Returns <StorageFormat>On<Engine>-like string.
     """
-    return f"{'Experimental' if IsExperimental.get() else ''}{StorageFormat.get()}On{Engine.get()}"
+    return f"{StorageFormat.get()}On{Engine.get()}"
 
 
 def instancer(_class: Callable[[], T]) -> T:


Code review comments:
diff_hunk: @@ -14,6 +14,7 @@
 """Collection of general utility functions, mostly for internal use."""
 
 import codecs
+import contextlib
path: modin/utils.py
text: @user1:
## Unused import

Import of 'contextlib' is not used.

[Show more details](https://github.com/modin-project/modin/security/code-scanning/1147)