FEAT-#4931: Create a query compiler that can connect to a service

.github/workflows/ci.yml

@@ -129,6 +129,10 @@ jobs:
       - run: python scripts/doc_checker.py modin/core/storage_formats/base
       - run: python scripts/doc_checker.py modin/experimental/core/storage_formats/pyarrow
       - run: python scripts/doc_checker.py modin/core/storage_formats/pandas
+      - run: |
+          python scripts/doc_checker.py modin/core/execution/client/container.py
+          python scripts/doc_checker.py modin/core/execution/client/io.py
+          python scripts/doc_checker.py modin/core/execution/client/query_compiler.py
       - run: |
           python scripts/doc_checker.py \
             modin/experimental/core/execution/native/implementations/hdk_on_native/dataframe \
@@ -335,7 +339,7 @@ jobs:
         shell: bash -l {0}
     strategy:
       matrix:
-        execution: [BaseOnPython]
+        execution: [BaseOnPython, Client]
     env:
       MODIN_TEST_DATASET_SIZE: "small"
     name: Test ${{ matrix.execution }} execution, Python 3.8
@@ -368,6 +372,8 @@ jobs:
       - name: Install HDF5
         run: sudo apt update && sudo apt install -y libhdf5-dev
       - run: pytest modin/experimental/xgboost/test/test_default.py --execution=${{ matrix.execution }}
+        # Client execution doesn't need to work with xgboost
+        if: matrix.execution != 'Client'      
       - run: python -m pytest -n 2 modin/test/storage_formats/base/test_internals.py --execution=${{ matrix.execution }}
       - run: pytest -n 2 modin/pandas/test/dataframe/test_binary.py --execution=${{ matrix.execution }}
       - run: pytest -n 2 modin/pandas/test/dataframe/test_default.py --execution=${{ matrix.execution }}
@@ -379,12 +385,22 @@ jobs:
       - run: pytest -n 2 modin/pandas/test/dataframe/test_udf.py --execution=${{ matrix.execution }}
       - run: pytest -n 2 modin/pandas/test/dataframe/test_window.py --execution=${{ matrix.execution }}
       - run: pytest -n 2 modin/pandas/test/dataframe/test_pickle.py --execution=${{ matrix.execution }}
+        # Client execution dosen't need to pickle modin.pandas objects.
+        if: matrix.execution != 'Client'      
       - run: python -m pytest -n 2 modin/pandas/test/test_series.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_rolling.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_concat.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_groupby.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_reshape.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_general.py --execution=${{ matrix.execution }}
+      - name: Test I/O
+        # note that if test dataset size is small like for the other tests in
+        # this job, the tests fail.
+        run: |
+          MODIN_TEST_DATASET_SIZE=NORMAL python -m pytest modin/pandas/test/test_io.py::TestCsv --execution=${{ matrix.execution }}
+          MODIN_TEST_DATASET_SIZE=NORMAL python -m pytest modin/pandas/test/test_io.py::TestSql --execution=${{ matrix.execution }}
+        # Client has to be able to to do CSV and SQL I/O.
+        if: matrix.execution == 'Client'
       - uses: codecov/codecov-action@v2
 
   test-hdk:

.github/workflows/push.yml

@@ -84,6 +84,8 @@ jobs:
       - name: Install HDF5
         run: sudo apt update && sudo apt install -y libhdf5-dev
       - run: pytest -n 2 modin/experimental/xgboost/test/test_default.py --execution=${{ matrix.execution }}
+        # Client execution doesn't need to work with xgboost
+        if: matrix.execution != 'Client'      
       - run: pytest -n 2 modin/pandas/test/dataframe/test_binary.py --execution=${{ matrix.execution }}
       - run: pytest -n 2 modin/pandas/test/dataframe/test_default.py --execution=${{ matrix.execution }}
       - run: pytest -n 2 modin/pandas/test/dataframe/test_indexing.py --execution=${{ matrix.execution }}
@@ -94,12 +96,20 @@ jobs:
       - run: pytest -n 2 modin/pandas/test/dataframe/test_udf.py --execution=${{ matrix.execution }}
       - run: pytest -n 2 modin/pandas/test/dataframe/test_window.py --execution=${{ matrix.execution }}
       - run: pytest -n 2 modin/pandas/test/dataframe/test_pickle.py --execution=${{ matrix.execution }}
+        # Client execution dosen't need to pickle modin.pandas objects.
+        if: matrix.execution != 'Client'      
       - run: python -m pytest -n 2 modin/pandas/test/test_series.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_rolling.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_concat.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_groupby.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_reshape.py --execution=${{ matrix.execution }}
       - run: python -m pytest -n 2 modin/pandas/test/test_general.py --execution=${{ matrix.execution }}
+      - name: I/O tests
+        run: |
+          python -m pytest modin/pandas/test/test_io.py::TestCsv --execution=${{ matrix.execution }}
+          python -m pytest modin/pandas/test/test_io.py::TestSql --execution=${{ matrix.execution }}
+        # Client has to be able to to CSV and SQL I/O.
+        if: matrix.execution == 'Client'      
       - uses: codecov/codecov-action@v2
 
   test-hdk:

modin/config/envvars.py

@@ -75,7 +75,7 @@ class Engine(EnvironmentVariable, type=str):
     """Distribution engine to run queries by."""
 
     varname = "MODIN_ENGINE"
-    choices = ("Ray", "Dask", "Python", "Native")
+    choices = ("Ray", "Dask", "Python", "Native", "Client")
 
     @classmethod
     def _get_default(cls) -> str:
@@ -131,17 +131,18 @@ def _get_default(cls) -> str:
             pass
         else:
             return "Native"
-        raise ImportError(
-            "Please refer to installation documentation page to install an engine"
-        )
+
+        # If we can't import any other engines we should go ahead and default to Python being
+        # the default backend engine.
+        return "Python"
 
 
 class StorageFormat(EnvironmentVariable, type=str):
     """Engine to run on a single node of distribution."""
 
     varname = "MODIN_STORAGE_FORMAT"
     default = "Pandas"
-    choices = ("Pandas", "Hdk", "Pyarrow", "Cudf")
+    choices = ("Pandas", "Hdk", "Pyarrow", "Cudf", "")
 
 
 class IsExperimental(EnvironmentVariable, type=bool):

modin/conftest.py

@@ -22,6 +22,7 @@
 import numpy as np
 import shutil
 from typing import Optional
+from uuid import uuid4
 
 assert (
     "modin.utils" not in sys.modules
@@ -46,12 +47,17 @@ def _saving_make_api_url(token, _make_api_url=modin.utils._make_api_url):
 import modin  # noqa: E402
 import modin.config  # noqa: E402
 from modin.config import IsExperimental, TestRayClient  # noqa: E402
-import uuid  # noqa: E402
 
 from modin.core.storage_formats import (  # noqa: E402
     PandasQueryCompiler,
     BaseQueryCompiler,
 )
+from modin.core.execution.client.container import (  # noqa: E402
+    ForwardingQueryCompilerContainer,
+)
+from modin.core.execution.python.implementations.pandas_on_python.dataframe.dataframe import (  # noqa: E402
+    PandasOnPythonDataframe,
+)
 from modin.core.execution.python.implementations.pandas_on_python.io import (  # noqa: E402
     PandasOnPythonIO,
 )
@@ -63,6 +69,7 @@ def _saving_make_api_url(token, _make_api_url=modin.utils._make_api_url):
     make_default_file,
     teardown_test_files,
     NROWS,
+    default_to_pandas_ignore_string,
 )
 
 
@@ -223,9 +230,6 @@ def __iter__(self):
     os.environ = orig_env
 
 
-BASE_EXECUTION_NAME = "BaseOnPython"
-
-
 class TestQC(BaseQueryCompiler):
     def __init__(self, modin_frame):
         self._modin_frame = modin_frame
@@ -269,16 +273,57 @@ def prepare(cls):
         cls.io_cls = BaseOnPythonIO
 
 
-def set_base_execution(name=BASE_EXECUTION_NAME):
-    setattr(factories, f"{name}Factory", BaseOnPythonFactory)
-    modin.set_execution(engine="python", storage_format=name.split("On")[0])
+def set_base_on_python_execution():
+    factories.BaseOnPythonFactory = BaseOnPythonFactory
+    modin.set_execution(engine="python", storage_format="Base")
+
+
+class ClientFactory(factories.BaseFactory):
+    @classmethod
+    def prepare(cls):
+        # Can't always import ClientIO, because it uses NoDefault, which
+        # is not available on older pandas.
+        from modin.core.execution.client.io import ClientIO
+
+        cls.io_cls = ClientIO
+
+
+def set_client_execution():
+    # Can't always import ClientQueryCompiler, because it uses NoDefault, which
+    # is not available on older pandas. ClientIO also uses ClientQueryCompiler.
+
+    from modin.core.execution.client.query_compiler import ClientQueryCompiler
+    from modin.core.execution.client.io import ClientIO
+
+    class TestClientQueryCompiler(ClientQueryCompiler):
+        @classmethod
+        def from_pandas(cls, df, data_cls):
+            return cls(
+                cls._service.add_query_compiler(TestQC.from_pandas(df, data_cls))
+            )
+
+        def default_to_pandas(self, pandas_op, *args, **kwargs):
+            result = self._service.default_to_pandas(
+                self._id, pandas_op, *args, **kwargs
+            )
+            if result.result_is_qc_id:
+                return self.__constructor__(result.result)
+            return result.result
+
+    service = ForwardingQueryCompilerContainer(BaseQueryCompiler, BaseOnPythonIO)
+    ClientQueryCompiler.set_server_connection(service)
+    ClientIO.query_compiler_cls = TestClientQueryCompiler
+    ClientIO.set_server_connection(service)
+    ClientIO.frame_cls = PandasOnPythonDataframe
+    factories.ClientFactory = ClientFactory
+    modin.set_execution(engine="Client", storage_format="")
 
 
 @pytest.fixture(scope="function")
 def get_unique_base_execution():
     """Setup unique execution for a single function and yield its QueryCompiler that's suitable for inplace modifications."""
     # It's better to use decimal IDs rather than hex ones due to factory names formatting
-    execution_id = int(uuid.uuid4().hex, 16)
+    execution_id = int(uuid4().hex, 16)
     format_name = f"Base{execution_id}"
     engine_name = "Python"
     execution_name = f"{format_name}On{engine_name}"
@@ -319,11 +364,12 @@ def pytest_configure(config):
     if execution is None:
         return
 
-    if execution == BASE_EXECUTION_NAME:
-        set_base_execution(BASE_EXECUTION_NAME)
-        config.addinivalue_line(
-            "filterwarnings", "default:.*defaulting to pandas.*:UserWarning"
-        )
+    if execution == "BaseOnPython":
+        set_base_on_python_execution()
+        config.addinivalue_line("filterwarnings", default_to_pandas_ignore_string)
+    elif execution == "Client":
+        set_client_execution()
+        config.addinivalue_line("filterwarnings", default_to_pandas_ignore_string)
     else:
         partition, engine = execution.split("On")
         modin.set_execution(engine=engine, storage_format=partition)