FEAT-#4605: Add native query compiler

.github/workflows/ci.yml

@@ -214,12 +214,16 @@ jobs:
       unidist: ${{ steps.filter.outputs.unidist }}
       engines: ${{ steps.engines.outputs.engines }}
       experimental: ${{ steps.experimental.outputs.experimental }}
+      test-native-dataframe-mode: ${{ steps.filter.outputs.test-native-dataframe-mode }}
     steps:
     - uses: actions/checkout@v4
     - uses: dorny/paths-filter@v3
       id: filter
       with:
         filters: |
+          test-native-dataframe-mode:
+            - 'modin/core/storage_formats/pandas/native_query_compiler.py'
+            - 'modin/core/storage_formats/base/query_compiler.py'
           shared: &shared
             - 'modin/core/execution/dispatching/**'
           ray:
@@ -631,6 +635,37 @@ jobs:
           python-version: ${{matrix.python-version}}
       - run: python -m pytest modin/tests/experimental/spreadsheet/test_general.py
 
+  test-native-dataframe-mode:
+    needs: [ lint-flake8, execution-filter]
+    if: ${{ needs.execution-filter.outputs.test-native-dataframe-mode == 'true' }}
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        shell: bash -l {0}
+    strategy:
+      matrix:
+        python-version: ["3.9"]
+    env:
+      MODIN_NATIVE_DATAFRAME_MODE: "Pandas"
+    name: test-native-dataframe-mode python ${{matrix.python-version}})
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/mamba-env
+        with:
+          environment-file: environment-dev.yml
+          python-version: ${{matrix.python-version}}
+      - run: python -m pytest modin/tests/pandas/dataframe/test_binary.py
+      - run: python -m pytest modin/tests/pandas/dataframe/test_default.py
+      - run: python -m pytest modin/tests/pandas/dataframe/test_indexing.py
+      - run: python -m pytest modin/tests/pandas/dataframe/test_iter.py
+      - run: python -m pytest modin/tests/pandas/dataframe/test_join_sort.py
+      - run: python -m pytest modin/tests/pandas/dataframe/test_map_metadata.py
+      - run: python -m pytest modin/tests/pandas/dataframe/test_pickle.py
+      - run: python -m pytest modin/tests/pandas/dataframe/test_reduce.py
+      - run: python -m pytest modin/tests/pandas/dataframe/test_udf.py
+      - run: python -m pytest modin/tests/pandas/dataframe/test_window.py
+      - uses: ./.github/actions/upload-coverage
+
   merge-coverage-artifacts:
     needs: [test-internals, test-api-and-no-engine, test-defaults, test-all-unidist, test-all, test-experimental, test-sanity]
     if: always()  # we need to run it regardless of some job being skipped, like in PR

modin/config/__init__.py

@@ -39,6 +39,7 @@
     MinPartitionSize,
     MinRowPartitionSize,
     ModinNumpy,
+    NativeDataframeMode,
     NPartitions,
     PersistentPickle,
     ProgressBar,
@@ -68,6 +69,7 @@
     "CpuCount",
     "GpuCount",
     "Memory",
+    "NativeDataframeMode",
     # Ray specific
     "IsRayCluster",
     "RayRedisAddress",

modin/config/envvars.py

@@ -913,4 +913,26 @@ def _check_vars() -> None:
         )
 
 
+class NativeDataframeMode(EnvironmentVariable, type=str):
+    """
+    Configures the query compiler to process Modin data.
+
+    When this config is set to ``Default``, ``PandasQueryCompiler`` is used,
+    which leads to Modin executing dataframes in distributed fashion.
+    When set to a string (e.g., ``pandas``), ``NativeQueryCompiler`` is used,
+    which handles the dataframes without distributing,
+    falling back to native library functions (e.g., ``pandas``).
+
+    This could be beneficial for handling relatively small dataframes
+    without involving additional overhead of communication between processes.
+    """
+
+    varname = "MODIN_NATIVE_DATAFRAME_MODE"
+    choices = (
+        "Default",
+        "Pandas",
+    )
+    default = "Default"
+
+
 _check_vars()

modin/core/execution/dispatching/factories/factories.py

@@ -26,7 +26,9 @@
 import pandas
 from pandas.util._decorators import doc
 
+from modin.config import NativeDataframeMode
 from modin.core.io import BaseIO
+from modin.core.storage_formats.pandas.native_query_compiler import NativeQueryCompiler
 from modin.utils import get_current_execution
 
 _doc_abstract_factory_class = """
@@ -168,6 +170,9 @@
         method="io.from_pandas",
     )
     def _from_pandas(cls, df):
+        if NativeDataframeMode.get() == "Pandas":
+            df_copy = df.copy()
+            return NativeQueryCompiler(df_copy)
         return cls.io_cls.from_pandas(df)
 
     @classmethod