Repository: modin-project/modin (python)

Issue:
Need the ability to configure NativeQueryCompiler with environment variables
We have several class variables in the native query compiler which should be configurable.

Pull request description:
FEAT-#7540: Ability to configure NativeQueryCompiler AutoSwitch Settings
Add the ability to change basic thresholds on the native query compiler

<!--
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. -->

- [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 #7540? <!-- 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 -->


Patch:
diff --git a/modin/config/__init__.py b/modin/config/__init__.py
index 6a258733..aeaadd95 100644
--- a/modin/config/__init__.py
+++ b/modin/config/__init__.py
@@ -44,6 +44,8 @@ from modin.config.envvars import (
     MinPartitionSize,
     MinRowPartitionSize,
     ModinNumpy,
+    NativePandasMaxRows,
+    NativePandasTransferThreshold,
     NPartitions,
     PersistentPickle,
     ProgressBar,
@@ -85,6 +87,9 @@ __all__ = [
     "LazyExecution",
     # Dask specific
     "DaskThreadsPerWorker",
+    # Native Pandas Specific
+    "NativePandasMaxRows",
+    "NativePandasTransferThreshold",
     # Partitioning
     "NPartitions",
     "MinPartitionSize",
diff --git a/modin/config/envvars.py b/modin/config/envvars.py
index 97c99f40..27fbda6e 100644
--- a/modin/config/envvars.py
+++ b/modin/config/envvars.py
@@ -1305,6 +1305,27 @@ class DaskThreadsPerWorker(EnvironmentVariable, type=int):
     default = 1
 
 
+class NativePandasMaxRows(EnvironmentVariable, type=int):
+    """
+    Maximum number of rows which can be processed using local, native, pandas.
+    """
+
+    varname = "MODIN_NATIVE_MAX_ROWS"
+    default = 10_000_000
+
+
+class NativePandasTransferThreshold(EnvironmentVariable, type=int):
+    """
+    Targeted max number of dataframe rows which should be transferred between engines.
+
+    This is often the same value as MODIN_NATIVE_MAX_ROWS but it can be independently
+    set to change how transfer costs are considered.
+    """
+
+    varname = "MODIN_NATIVE_MAX_XFER_ROWS"
+    default = 10_000_000
+
+
 class DynamicPartitioning(EnvironmentVariable, type=bool):
     """
     Set to true to use Modin's dynamic-partitioning implementation where possible.
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
index a1375677..0ac79bb2 100644
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -356,13 +356,13 @@ class BaseQueryCompiler(
         """
         if isinstance(self, other_qc_type):
             return QCCoercionCost.COST_ZERO
-        if self._TRANSFER_THRESHOLD <= 0:
+        if self.__class__._transfer_threshold() <= 0:
             return QCCoercionCost.COST_ZERO
         cost = int(
             (
                 QCCoercionCost.COST_IMPOSSIBLE
                 * self._max_shape()[0]
-                / self._TRANSFER_THRESHOLD
+                / self.__class__._transfer_threshold()
             )
         )
         if cost > QCCoercionCost.COST_IMPOSSIBLE:
@@ -447,7 +447,7 @@ class BaseQueryCompiler(
         return self._stay_cost_rows(
             self._max_shape()[0],
             self._OPERATION_PER_ROW_OVERHEAD,
-            self._MAX_SIZE_THIS_ENGINE_CAN_HANDLE,
+            self.__class__._engine_max_size(),
             self._OPERATION_INITIALIZATION_OVERHEAD,
         )
 
@@ -498,10 +498,18 @@ class BaseQueryCompiler(
         return cls._stay_cost_rows(
             row_count,
             cls._OPERATION_PER_ROW_OVERHEAD,
-            cls._MAX_SIZE_THIS_ENGINE_CAN_HANDLE,
+            cls._engine_max_size(),
             cls._OPERATION_INITIALIZATION_OVERHEAD,
         )
 
+    @classmethod
+    def _engine_max_size(cls):
+        return cls._MAX_SIZE_THIS_ENGINE_CAN_HANDLE
+
+    @classmethod
+    def _transfer_threshold(cls):
+        return cls._TRANSFER_THRESHOLD
+
     @disable_logging
     def max_cost(self) -> int:
         """
diff --git a/modin/core/storage_formats/pandas/native_query_compiler.py b/modin/core/storage_formats/pandas/native_query_compiler.py
index 74f8b694..ad17a115 100644
--- a/modin/core/storage_formats/pandas/native_query_compiler.py
+++ b/modin/core/storage_formats/pandas/native_query_compiler.py
@@ -23,12 +23,11 @@ from typing import Optional
 import pandas
 from pandas.core.dtypes.common import is_scalar
 
+from modin.config.envvars import NativePandasMaxRows, NativePandasTransferThreshold
 from modin.core.dataframe.base.interchange.dataframe_protocol.dataframe import (
     ProtocolDataframe,
 )
-from modin.core.storage_formats.base.query_compiler import (
-    BaseQueryCompiler,
-)
+from modin.core.storage_formats.base.query_compiler import BaseQueryCompiler
 from modin.utils import _inherit_docstrings
 
 _NO_REPARTITION_ON_NATIVE_EXECUTION_EXCEPTION_MESSAGE = (
@@ -92,10 +91,8 @@ class NativeQueryCompiler(BaseQueryCompiler):
         The pandas frame to query with the compiled queries.
     """
 
-    _MAX_SIZE_THIS_ENGINE_CAN_HANDLE = 10_000_000
     _OPERATION_INITIALIZATION_OVERHEAD = 0
     _OPERATION_PER_ROW_OVERHEAD = 0
-    _TRANSFER_THRESHOLD = 10_000_000
 
     _modin_frame: pandas.DataFrame
     _should_warn_on_default_to_pandas: bool = False
@@ -212,8 +209,21 @@ class NativeQueryCompiler(BaseQueryCompiler):
     def finalize(self):
         return
 
-    # Dataframe interchange protocol
+    @classmethod
+    def _engine_max_size(cls):
+        # do not return the custom configuration for sub-classes
+        if cls == NativeQueryCompiler:
+            return NativePandasMaxRows.get()
+        return cls._MAX_SIZE_THIS_ENGINE_CAN_HANDLE
 
+    @classmethod
+    def _transfer_threshold(cls):
+        # do not return the custom configuration for sub-classes
+        if cls == NativeQueryCompiler:
+            return NativePandasTransferThreshold.get()
+        return cls._TRANSFER_THRESHOLD
+
+    # Dataframe interchange protocol
     def to_interchange_dataframe(
         self, nan_as_null: bool = False, allow_copy: bool = True
     ):


Test patch:
diff --git a/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py b/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py
index cfef3ae7..c5bb52d5 100644
--- a/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py
+++ b/modin/tests/pandas/native_df_interoperability/test_compiler_caster.py
@@ -24,7 +24,13 @@ from pytest import param
 
 import modin.pandas as pd
 from modin.config import context as config_context
-from modin.config.envvars import Backend, Engine, Execution
+from modin.config.envvars import (
+    Backend,
+    Engine,
+    Execution,
+    NativePandasMaxRows,
+    NativePandasTransferThreshold,
+)
 from modin.core.execution.dispatching.factories import factories
 from modin.core.execution.dispatching.factories.factories import BaseFactory
 from modin.core.io.io import BaseIO
@@ -1271,3 +1277,34 @@ def test_second_init_only_calls_from_pandas_once_github_issue_7559():
         ) as mock_from_pandas:
             pd.DataFrame([1])
             mock_from_pandas.assert_called_once()
+
+
+def test_native_config():
+    qc = NativeQueryCompiler(pandas.DataFrame([0, 1, 2]))
+
+    # Native Query Compiler gets a special configuration
+    assert qc._TRANSFER_THRESHOLD == 0
+    assert qc._transfer_threshold() == NativePandasTransferThreshold.get()
+    assert qc._MAX_SIZE_THIS_ENGINE_CAN_HANDLE == 1
+    assert qc._engine_max_size() == NativePandasMaxRows.get()
+
+    oldmax = qc._engine_max_size()
+    oldthresh = qc._transfer_threshold()
+
+    with config_context(NativePandasMaxRows=123, NativePandasTransferThreshold=321):
+        qc2 = NativeQueryCompiler(pandas.DataFrame([0, 1, 2]))
+        assert qc2._transfer_threshold() == 321
+        assert qc2._engine_max_size() == 123
+        assert qc._engine_max_size() == 123
+        assert qc._transfer_threshold() == 321
+
+        # sub class configuration is unchanged
+        class AQC(NativeQueryCompiler):
+            pass
+
+        subqc = AQC(pandas.DataFrame([0, 1, 2]))
+        assert subqc._TRANSFER_THRESHOLD == 0
+        assert subqc._MAX_SIZE_THIS_ENGINE_CAN_HANDLE == 1
+
+    assert qc._engine_max_size() == oldmax
+    assert qc._transfer_threshold() == oldthresh