Repository: modin-project/modin (python)

Issue:
Track engine and storage format at dataframe & series level
Currently, `Engine` and `StorageFormat` determine the execution for new Modin objects.

We should track and propagate these values for individual dataframes, so that users can ultimately view them and change them.

Pull request description:
REFACTOR-#7427: Require query compilers to expose engine and storage format.
Make `engine` and `storage_format` required read-only properties for query compiler. We track `engine` and `storage_format` as follows:
- `TestQc` is always on `engine="Base",storage_format="Python"`
- `PandasQueryCompiler` gets its engine and storage format from its `_modin_frame`.
- `NativeQueryCompiler` is always on `engine="Base",storage_format="Native"`.

For testing, this commit adds a step to `eval_general` that checks that if `operation` takes a dataframe or series as its first input and produces a dataframe or series output, then it preserves  `engine` and `storage_format`. We can disable this step in rare cases where the engine changes in erroneous or unpredictable ways. Also, this commit adds a step to `eval_io` that checks that newly created dataframes and series have an engine and storage format that match the default `Engine` and `StorageFormat`.

- [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 #7427
- [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 -->


Patch:
diff --git a/modin/core/dataframe/pandas/dataframe/dataframe.py b/modin/core/dataframe/pandas/dataframe/dataframe.py
index 043de08f..88869f12 100644
--- a/modin/core/dataframe/pandas/dataframe/dataframe.py
+++ b/modin/core/dataframe/pandas/dataframe/dataframe.py
@@ -22,6 +22,7 @@ from __future__ import annotations
 
 import datetime
 import re
+from abc import ABC, abstractmethod
 from functools import cached_property
 from typing import TYPE_CHECKING, Callable, Dict, Hashable, List, Optional, Union
 
@@ -33,7 +34,6 @@ from pandas.core.dtypes.common import is_dtype_equal, is_list_like, is_numeric_d
 from pandas.core.indexes.api import Index, RangeIndex
 
 from modin.config import (
-    Engine,
     IsRayCluster,
     MinColumnPartitionSize,
     MinRowPartitionSize,
@@ -80,7 +80,7 @@ if TYPE_CHECKING:
 
 
 class PandasDataframe(
-    ClassLogger, modin_layer="CORE-DATAFRAME", log_level=LogLevel.DEBUG
+    ABC, ClassLogger, modin_layer="CORE-DATAFRAME", log_level=LogLevel.DEBUG
 ):
     """
     An abstract class that represents the parent class for any pandas storage format dataframe class.
@@ -122,6 +122,31 @@ class PandasDataframe(
     _dtypes: Optional[ModinDtypes] = None
     _pandas_backend: Optional[str] = None
 
+    @property
+    def storage_format(self) -> str:
+        """
+        The storage format for this frame's data.
+
+        Returns
+        -------
+        str
+            The storage format.
+        """
+        return "Pandas"
+
+    @property
+    @abstractmethod
+    def engine(self) -> str:
+        """
+        The engine for this frame.
+
+        Returns
+        -------
+        str
+            The engine.
+        """
+        pass
+
     @cached_property
     def __constructor__(self) -> type[PandasDataframe]:
         """
@@ -1707,7 +1732,7 @@ class PandasDataframe(
                         new_dtypes = self_dtypes.copy()
                     # Update the new dtype series to the proper pandas dtype
                     new_dtype = pandas.api.types.pandas_dtype(dtype)
-                    if Engine.get() == "Dask" and hasattr(dtype, "_is_materialized"):
+                    if self.engine == "Dask" and hasattr(dtype, "_is_materialized"):
                         # FIXME: https://github.com/dask/distributed/issues/8585
                         _ = dtype._materialize_categories()
 
@@ -1736,7 +1761,7 @@ class PandasDataframe(
             if not (col_dtypes == self_dtypes).all():
                 new_dtypes = self_dtypes.copy()
                 new_dtype = pandas.api.types.pandas_dtype(col_dtypes)
-                if Engine.get() == "Dask" and hasattr(new_dtype, "_is_materialized"):
+                if self.engine == "Dask" and hasattr(new_dtype, "_is_materialized"):
                     # FIXME: https://github.com/dask/distributed/issues/8585
                     _ = new_dtype._materialize_categories()
                 if isinstance(new_dtype, pandas.CategoricalDtype):
diff --git a/modin/core/dataframe/pandas/metadata/dtypes.py b/modin/core/dataframe/pandas/metadata/dtypes.py
index 9220a2da..327d6434 100644
--- a/modin/core/dataframe/pandas/metadata/dtypes.py
+++ b/modin/core/dataframe/pandas/metadata/dtypes.py
@@ -1191,7 +1191,7 @@ class LazyProxyCategoricalDtype(pandas.CategoricalDtype):
 
 
 def get_categories_dtype(
-    cdt: Union[LazyProxyCategoricalDtype, pandas.CategoricalDtype]
+    cdt: Union[LazyProxyCategoricalDtype, pandas.CategoricalDtype],
 ) -> DtypeObj:
     """
     Get the categories dtype.
diff --git a/modin/core/execution/dask/implementations/pandas_on_dask/dataframe/dataframe.py b/modin/core/execution/dask/implementations/pandas_on_dask/dataframe/dataframe.py
index 5e4598d0..034b24bc 100644
--- a/modin/core/execution/dask/implementations/pandas_on_dask/dataframe/dataframe.py
+++ b/modin/core/execution/dask/implementations/pandas_on_dask/dataframe/dataframe.py
@@ -14,6 +14,7 @@
 """Module houses class that implements ``PandasDataframe``."""
 
 from modin.core.dataframe.pandas.dataframe.dataframe import PandasDataframe
+from modin.utils import _inherit_docstrings
 
 from ..partitioning.partition_manager import PandasOnDaskDataframePartitionManager
 
@@ -66,3 +67,8 @@ class PandasOnDaskDataframe(PandasDataframe):
 
         address = default_client().scheduler_info()["address"]
         return self.reconnect, (address, self.__dict__)
+
+    @property
+    @_inherit_docstrings(PandasDataframe.engine)
+    def engine(self) -> str:
+        return "Dask"
diff --git a/modin/core/execution/dispatching/factories/factories.py b/modin/core/execution/dispatching/factories/factories.py
index ba6d1eef..16acd9ff 100644
--- a/modin/core/execution/dispatching/factories/factories.py
+++ b/modin/core/execution/dispatching/factories/factories.py
@@ -629,6 +629,8 @@ class BaseFactory(object):
         **kwargs : kwargs
             Arguments to the writer method.
         """
+        # TODO(https://github.com/modin-project/modin/issues/7429): Use
+        # frame-level execution instead of the global, default execution.
         current_execution = get_current_execution()
         if current_execution not in supported_executions:
             raise NotImplementedError(
diff --git a/modin/core/execution/modin_aqp.py b/modin/core/execution/modin_aqp.py
index f1ba7bdc..c2d85ccc 100644
--- a/modin/core/execution/modin_aqp.py
+++ b/modin/core/execution/modin_aqp.py
@@ -91,6 +91,8 @@ def call_progress_bar(result_parts, line_no):
 
     threading.Thread(target=_show_time_updates, args=(progress_bars[pbar_id],)).start()
 
+    # TODO(https://github.com/modin-project/modin/issues/7429): Use
+    # frame-level engine config.
     modin_engine = Engine.get()
     engine_wrapper = None
     if modin_engine == "Ray":
diff --git a/modin/core/execution/python/implementations/pandas_on_python/dataframe/dataframe.py b/modin/core/execution/python/implementations/pandas_on_python/dataframe/dataframe.py
index 0e2bc70d..e09b5152 100644
--- a/modin/core/execution/python/implementations/pandas_on_python/dataframe/dataframe.py
+++ b/modin/core/execution/python/implementations/pandas_on_python/dataframe/dataframe.py
@@ -18,6 +18,7 @@ Module contains class ``PandasOnPythonDataframe``.
 """
 
 from modin.core.dataframe.pandas.dataframe.dataframe import PandasDataframe
+from modin.utils import _inherit_docstrings
 
 from ..partitioning.partition_manager import PandasOnPythonDataframePartitionManager
 
@@ -50,3 +51,8 @@ class PandasOnPythonDataframe(PandasDataframe):
     """
 
     _partition_mgr_cls = PandasOnPythonDataframePartitionManager
+
+    @property
+    @_inherit_docstrings(PandasDataframe.engine)
+    def engine(self) -> str:
+        return "Python"
diff --git a/modin/core/execution/ray/implementations/pandas_on_ray/dataframe/dataframe.py b/modin/core/execution/ray/implementations/pandas_on_ray/dataframe/dataframe.py
index 373a84ec..691189e4 100644
--- a/modin/core/execution/ray/implementations/pandas_on_ray/dataframe/dataframe.py
+++ b/modin/core/execution/ray/implementations/pandas_on_ray/dataframe/dataframe.py
@@ -15,6 +15,7 @@
 
 from modin.core.dataframe.base.dataframe.utils import Axis
 from modin.core.dataframe.pandas.dataframe.dataframe import PandasDataframe
+from modin.utils import _inherit_docstrings
 
 from ..partitioning.partition_manager import PandasOnRayDataframePartitionManager
 
@@ -66,3 +67,8 @@ class PandasOnRayDataframe(PandasDataframe):
             dims = [part.width(False) for part in parts]
 
         return self._partition_mgr_cls.materialize_futures(dims)
+
+    @property
+    @_inherit_docstrings(PandasDataframe.engine)
+    def engine(self) -> str:
+        return "Ray"
diff --git a/modin/core/execution/unidist/implementations/pandas_on_unidist/dataframe/dataframe.py b/modin/core/execution/unidist/implementations/pandas_on_unidist/dataframe/dataframe.py
index 9adba6bc..d885b16d 100644
--- a/modin/core/execution/unidist/implementations/pandas_on_unidist/dataframe/dataframe.py
+++ b/modin/core/execution/unidist/implementations/pandas_on_unidist/dataframe/dataframe.py
@@ -14,6 +14,7 @@
 """Module houses class that implements ``PandasDataframe`` using unidist."""
 
 from modin.core.dataframe.pandas.dataframe.dataframe import PandasDataframe
+from modin.utils import _inherit_docstrings
 
 from ..partitioning.partition_manager import PandasOnUnidistDataframePartitionManager
 
@@ -47,3 +48,8 @@ class PandasOnUnidistDataframe(PandasDataframe):
     def support_materialization_in_worker_process(self) -> bool:
         # more details why this is not `True` in https://github.com/modin-project/modin/pull/6673
         return False
+
+    @property
+    @_inherit_docstrings(PandasDataframe.engine)
+    def engine(self) -> str:
+        return "Unidist"
diff --git a/modin/core/execution/utils.py b/modin/core/execution/utils.py
index 64c8bdc5..d9504e40 100644
--- a/modin/core/execution/utils.py
+++ b/modin/core/execution/utils.py
@@ -42,6 +42,9 @@ if "_MODIN_DOC_CHECKER_" in os.environ:
 
 # Check if the function already exists to avoid circular imports
 elif "remote_function" not in dir():
+    # TODO(https://github.com/modin-project/modin/issues/7429): Use
+    # frame-level engine config.
+
     from modin.config import Engine
 
     if Engine.get() == "Ray":
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
index 1b4284a1..43e07be4 100644
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -150,6 +150,32 @@ class BaseQueryCompiler(
     _modin_frame: PandasDataframe
     _shape_hint: Optional[str]
 
+    @property
+    @abc.abstractmethod
+    def storage_format(self) -> str:
+        """
+        The storage format for this query compiler.
+
+        Returns
+        -------
+        str
+            The storage format.
+        """
+        pass
+
+    @property
+    @abc.abstractmethod
+    def engine(self) -> str:
+        """
+        The engine for this query compiler.
+
+        Returns
+        -------
+        str
+            The engine.
+        """
+        pass
+
     def __wrap_in_qc(self, obj):
         """
         Wrap `obj` in query compiler.
diff --git a/modin/core/storage_formats/pandas/native_query_compiler.py b/modin/core/storage_formats/pandas/native_query_compiler.py
index 7660c05f..c724cfc9 100644
--- a/modin/core/storage_formats/pandas/native_query_compiler.py
+++ b/modin/core/storage_formats/pandas/native_query_compiler.py
@@ -598,6 +598,11 @@ class NativeQueryCompiler(BaseQueryCompiler, QueryCompilerCaster):
         self._modin_frame = pandas_frame
         self._shape_hint = shape_hint
 
+    storage_format = property(
+        lambda self: "Pandas", doc=BaseQueryCompiler.storage_format.__doc__
+    )
+    engine = property(lambda self: "Native", doc=BaseQueryCompiler.engine.__doc__)
+
     def execute(self):
         pass
 
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 656b636d..449eeecd 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -299,6 +299,9 @@ class PandasQueryCompiler(BaseQueryCompiler, QueryCompilerCaster):
         self._modin_frame = modin_frame
         self._shape_hint = shape_hint
 
+    storage_format = property(lambda self: self._modin_frame.storage_format)
+    engine = property(lambda self: self._modin_frame.engine)
+
     @property
     def lazy_row_labels(self):
         """
diff --git a/modin/error_message.py b/modin/error_message.py
index a31cd5da..f42f7810 100644
--- a/modin/error_message.py
+++ b/modin/error_message.py
@@ -55,6 +55,9 @@ class ErrorMessage(object):
 
     @classmethod
     def default_to_pandas(cls, message: str = "", reason: str = "") -> None:
+        # TODO(https://github.com/modin-project/modin/issues/7429): Use
+        # frame-level engine config.
+
         if message != "":
             execution_str = get_current_execution()
             message = (
diff --git a/modin/pandas/io.py b/modin/pandas/io.py
index ecd53935..c43ee436 100644
--- a/modin/pandas/io.py
+++ b/modin/pandas/io.py
@@ -1133,7 +1133,7 @@ def to_pandas(modin_obj: SupportsPublicToPandas) -> DataFrame | Series:
 
 
 def to_numpy(
-    modin_obj: Union[SupportsPrivateToNumPy, SupportsPublicToNumPy]
+    modin_obj: Union[SupportsPrivateToNumPy, SupportsPublicToNumPy],
 ) -> np.ndarray:
     """
     Convert a Modin object to a NumPy array.


Test patch:
diff --git a/modin/conftest.py b/modin/conftest.py
index 9a126883..cc7f87f3 100644
--- a/modin/conftest.py
+++ b/modin/conftest.py
@@ -165,6 +165,11 @@ class TestQC(BaseQueryCompiler):
     def __init__(self, modin_frame):
         self._modin_frame = modin_frame
 
+    storage_format = property(
+        lambda self: "Base", doc=BaseQueryCompiler.storage_format.__doc__
+    )
+    engine = property(lambda self: "Python", doc=BaseQueryCompiler.engine.__doc__)
+
     def finalize(self):
         self._modin_frame.finalize()
 
diff --git a/modin/tests/pandas/native_df_mode/test_indexing.py b/modin/tests/pandas/native_df_mode/test_indexing.py
index b4340263..4046f696 100644
--- a/modin/tests/pandas/native_df_mode/test_indexing.py
+++ b/modin/tests/pandas/native_df_mode/test_indexing.py
@@ -428,6 +428,10 @@ def test_setitem_on_empty_df(data, value, convert_to_series, new_col_id, df_mode
             "check_dtypes": not (len(pandas_df) == 0 and len(pandas_df.columns) != 0)
         },
         expected_exception=expected_exception,
+        check_for_execution_propagation=False,
+        no_check_for_execution_propagation_reason=(
+            "https://github.com/modin-project/modin/issues/7428"
+        ),
     )
 
 
diff --git a/modin/tests/pandas/test_io.py b/modin/tests/pandas/test_io.py
index bade22ef..94bf7418 100644
--- a/modin/tests/pandas/test_io.py
+++ b/modin/tests/pandas/test_io.py
@@ -69,7 +69,9 @@ from .utils import (
     parse_dates_values_by_id,
 )
 from .utils import test_data as utils_test_data
-from .utils import time_parsing_csv_path
+from .utils import (
+    time_parsing_csv_path,
+)
 
 if StorageFormat.get() == "Pandas":
     import modin.pandas as pd
diff --git a/modin/tests/pandas/utils.py b/modin/tests/pandas/utils.py
index 16f5a87c..1d80720b 100644
--- a/modin/tests/pandas/utils.py
+++ b/modin/tests/pandas/utils.py
@@ -45,6 +45,7 @@ from modin.config import (
     MinRowPartitionSize,
     NPartitions,
     RangePartitioning,
+    StorageFormat,
     TestDatasetSize,
     TrackFileLeaks,
 )
@@ -924,10 +925,19 @@ def eval_general(
     check_kwargs_callable=True,
     md_extra_kwargs=None,
     comparator_kwargs=None,
+    check_for_execution_propagation=True,
+    no_check_for_execution_propagation_reason=None,
     **kwargs,
 ):
     md_kwargs, pd_kwargs = {}, {}
 
+    if isinstance(modin_df, (pd.DataFrame, pd.Series)):
+        original_engine = modin_df._query_compiler.engine
+        original_storage_format = modin_df._query_compiler.storage_format
+    else:
+        original_engine = None
+        original_storage_format = None
+
     def execute_callable(fn, inplace=False, md_kwargs={}, pd_kwargs={}):
         try:
             pd_result = fn(pandas_df, **pd_kwargs)
@@ -1000,7 +1010,35 @@ def eval_general(
         operation, md_kwargs=md_kwargs, pd_kwargs=pd_kwargs, inplace=__inplace__
     )
     if values is not None:
-        comparator(*values, **(comparator_kwargs or {}))
+        assert isinstance(values, tuple) and len(values) == 2
+        modin_result, pandas_result = values
+        if (
+            isinstance(modin_result, (pd.DataFrame, pd.Series))
+            and original_engine is not None
+            and original_storage_format is not None
+        ):
+            if check_for_execution_propagation:
+                assert modin_result._query_compiler.engine == original_engine, (
+                    f"Result engine {modin_result._query_compiler.engine} does "
+                    + f"not match expected engine {original_engine}"
+                )
+                assert (
+                    modin_result._query_compiler.storage_format
+                    == original_storage_format
+                ), (
+                    "Result storage format "
+                    + f"{modin_result._query_compiler.storage_format} does "
+                    + f"not match expected storage format {original_storage_format}"
+                )
+            else:
+                assert (
+                    isinstance(no_check_for_execution_propagation_reason, str)
+                    and len(no_check_for_execution_propagation_reason) > 0
+                ), (
+                    "Must provide a reason for not expecting the operation to "
+                    + "propagate dataframe/series engine."
+                )
+        comparator(modin_result, pandas_result, **(comparator_kwargs or {}))
 
 
 def eval_io(
@@ -1042,6 +1080,12 @@ def eval_io(
         result = getattr(module, fn_name)(*args, **kwargs)
         if cast_to_str:
             result = result.astype(str)
+        if isinstance(result, (pd.DataFrame, pd.Series)):
+            # Input methods that return a dataframe, e.g. read_csv, should
+            # return a dataframe with engine and storage_format that match
+            # the default Engine and StorageFormat, respectively.
+            assert result._query_compiler.engine == Engine.get()
+            assert result._query_compiler.storage_format == StorageFormat.get()
         return result
 
     def call_eval_general():
diff --git a/modin/tests/test_executions_api.py b/modin/tests/test_executions_api.py
index cde0d7f3..8efd8c49 100644
--- a/modin/tests/test_executions_api.py
+++ b/modin/tests/test_executions_api.py
@@ -31,6 +31,8 @@ def test_base_abstract_methods():
         "default_to_pandas",
         "from_interchange_dataframe",
         "to_interchange_dataframe",
+        "engine",
+        "storage_format",
     ]
 
     not_implemented_methods = BASE_EXECUTION.__abstractmethods__.difference(