Repository: modin-project/modin
Pull request: FIX-#3368: support unsigned integers in OmniSci backend
<!--
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?

This PR introduces a native workaround for working with unsigned integers via OmniSci backend. Originally, OmniSci doesn't support unsigned integers, the PR adds a logic of upcasting such types to the compatible ones prior importing them to OmniSci.

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
- [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 #3368 <!-- 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 -->
- [x] added (Issue Number: PR title (PR Number)) and github username to release notes for next major release <!-- e.g. DOCS-#4077: Add release notes template to docs folder (#4078) -->


Problem statement:
RuntimeError: uint8 is not yet supported.
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**: `Ubuntu 20.04`
- **Modin version** (`modin.__version__`): `0.10.2` (from conda-forge)
- **Python version**: `3.9.6`
- **Code we can use to reproduce**:

```shell
conda install modin-omnisci -c conda-forge
MODIN_BACKEND=omnisci MODIN_EXPERIMENTAL=true python test.py
```

Test.py:
```python
import numpy as np
import modin.pandas as pd

test = np.array([1., 3., 5.])
pd.get_dummies(np.array([1., 3., 5.])).sum(axis=0)
```

### Describe the problem
<!-- Describe the problem clearly here. -->
This issue was found when running `PlastiCC` workload.

### Source code / logs
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->

```
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File ".../modin_on_omnisci_conda_forge/lib/python3.9/site-packages/modin/pandas/dataframe.py", line 2070, in sum
    data._query_compiler.sum(
  File ".../modin_on_omnisci_conda_forge/lib/python3.9/site-packages/modin/experimental/backends/omnisci/query_compiler.py", line 96, in method_wrapper
    return method(self, *args, **kwargs)
  File ".../modin_on_omnisci_conda_forge/lib/python3.9/site-packages/modin/experimental/backends/omnisci/query_compiler.py", line 361, in sum
    return self._agg("sum", **kwargs)
  File ".../modin_on_omnisci_conda_forge/lib/python3.9/site-packages/modin/experimental/backends/omnisci/query_compiler.py", line 96, in method_wrapper
    return method(self, *args, **kwargs)
  File ".../modin_on_omnisci_conda_forge/lib/python3.9/site-packages/modin/experimental/backends/omnisci/query_compiler.py", line 413, in _agg
    new_frame = new_frame._set_index(
  File ".../modin_on_omnisci_conda_forge/lib/python3.9/site-packages/modin/experimental/engines/omnisci_on_ray/frame/data.py", line 1931, in _set_index
    self._execute()
  File ".../modin_on_omnisci_conda_forge/lib/python3.9/site-packages/modin/experimental/engines/omnisci_on_ray/frame/data.py", line 1708, in _execute
    new_partitions = self._partition_mgr_cls.run_exec_plan(
  File ".../modin_on_omnisci_conda_forge/lib/python3.9/site-packages/modin/experimental/engines/omnisci_on_ray/frame/partition_manager.py", line 248, in run_exec_plan
    p.frame_id = omniSession.put_arrow_to_omnisci(obj)
  File ".../modin_on_omnisci_conda_forge/lib/python3.9/site-packages/modin/experimental/engines/omnisci_on_ray/frame/omnisci_worker.py", line 197, in put_arrow_to_omnisci
    cls._server.importArrowTable(name, table, fragment_size=fragment_size)
  File "dbe.pyx", line 207, in omniscidbe.PyDbEngine.importArrowTable
RuntimeError: uint8 is not yet supported.
```

Hints from issue discussion:
@dchigarev can you take a look?
As far as I know, OmniSci doesn't support uints for most operations.

cc @ienkovich , @fexolm 
it seems that PyDBE (not sure about OmniSci itself) does not support unsigned integers at all:

<details><summary>Reproducer</summary>

```python
import sys

sys.setdlopenflags(1 | 256)  # RTLD_LAZY+RTLD_GLOBAL
from dbe import PyDbEngine

import pyarrow as pa

server = PyDbEngine()

data = {"a": [1, 2, 3]}

dtypes = (
    "uint8", "uint16", "uint32", "uint64",
    "int8", "int16", "int32", "int64"
)
tables = (
    pa.Table.from_pydict(data, schema=pa.schema({"a": getattr(pa, dtype)()}))
    for dtype in dtypes
)

for dtype, table in zip(dtypes, tables):
    try:
        server.importArrowTable(dtype, table)
    except Exception as e:
        print(f"{dtype} import has failed: {e}")
    else:
        print(f"{dtype} has been imported successfully")
```

</details>

Output:
```
uint8 import has failed: uint8 is not yet supported.
uint16 import has failed: uint16 is not yet supported.
uint32 import has failed: uint32 is not yet supported.
uint64 import has failed: uint64 is not yet supported.
int8 has been imported successfully
int16 has been imported successfully
int32 has been imported successfully
int64 has been imported successfully

```

The simplest workaround that I could find for this issue is the following:
```python
import numpy as np
import modin.pandas as pd
import pandas

uint8_dummies = pd.get_dummies(np.array([1., 3., 5.]))
# Modin's astype will trigger the data import to OmniSci as well, so defaulting to pandas
valid_dummies = uint8_dummies._default_to_pandas(pandas.DataFrame.astype, "int8")
# Workaround for Modin issue #3370
valid_dummies.columns = valid_dummies.columns.astype("str")
result = valid_dummies.sum(axis=0)
print(result)
```

@dchigarev What about workaround performance? Maybe it's faster to default to pandas with appropriate warning?
For reference, this is a patch for plasticc benchmark example that makes it work:
```patch
diff --git a/examples/docker/modin-omnisci/plasticc-omnisci.py b/examples/docker/modin-omnisci/plasticc-omnisci.py
index 958a524d..6a505b51 100644
--- a/examples/docker/modin-omnisci/plasticc-omnisci.py
+++ b/examples/docker/modin-omnisci/plasticc-omnisci.py
@@ -16,6 +16,7 @@ import time
 from collections import OrderedDict
 from functools import partial
 import modin.pandas as pd
+import pandas
 from modin.experimental.engines.omnisci_on_ray.frame.omnisci_worker import OmnisciServer

 import numpy as np
@@ -118,10 +119,12 @@ def multi_weighted_logloss(y_true, y_preds, classes, class_weights):
     """
     y_p = y_preds.reshape(y_true.shape[0], len(classes), order="F")
     y_ohe = pd.get_dummies(y_true)
+    valid_y_ohe = y_ohe._default_to_pandas(pandas.DataFrame.astype, "int8")
+    valid_y_ohe.columns = valid_y_ohe.columns.astype("str")
     y_p = np.clip(a=y_p, a_min=1e-15, a_max=1 - 1e-15)
     y_p_log = np.log(y_p)
     y_log_ones = np.sum(y_ohe.values * y_p_log, axis=0)
-    nb_pos = y_ohe.sum(axis=0).values.astype(float)
+    nb_pos = valid_y_ohe.sum(axis=0).values.astype(float)
     class_arr = np.array([class_weights[k] for k in sorted(class_weights.keys())])
     y_w = y_log_ones * class_arr / nb_pos
```
> As far as I know, OmniSci doesn't support uints for most operations.
> 
> cc @ienkovich , @fexolm

that's right. 
@gshimansky  How does benchmark source code changes affect performance?
BTW, since `get_dummies` itself already fallbacks to pandas, we can just put all of the steps of the presented workaround into a single default-to-pandas call instead of doing it multiple times and waste time on unnecessary conversions from modin to pandas and vice versa:
```python
import numpy as np
import modin.pandas as pd
from modin.utils import try_cast_to_pandas
import pandas

test = np.array([1., 3., 5.])
# Calling `pandas.get_dummies` instead of modin to get pandas object in the result:
pandas_uint_dummies = pandas.get_dummies(try_cast_to_pandas(test))
pandas_uint_dummies = pandas_uint_dummies.astype("int8")
# Workaround for Modin issue #3370
pandas_uint_dummies.columns = pandas_uint_dummies.columns.astype("str")

# Converting valid for OmniSci result back to modin
modin_valid_dummies = pd.DataFrame(pandas_uint_dummies)
print(modin_valid_dummies.sum())

```

Using this workaround appears to be twice faster than the one [suggested by me before](https://github.com/modin-project/modin/issues/3368#issuecomment-904569920).

<details><summary>Perf comparison of these workarounds</summary>

```python
from modin.config import Backend, Engine

Backend.put("Omnisci")
Engine.put("Native")

import modin.pandas as pd
from modin.utils import try_cast_to_pandas
import pandas
import timeit


def uint_get_dummies_only_pandas_workaround(df): # timeit(number=10) -> 3.15s
    """This function uses only pandas."""
    df = try_cast_to_pandas(df)
    # Calling `pandas.get_dummies` instead of modin to get pandas object in the result:
    result = pandas.get_dummies(df).astype("int8")
    # Workaround for Modin issue #3370
    result.columns = result.columns.astype("str")
    # Converting valid for OmniSci result back to modin:
    result = pd.DataFrame(result)
    return result

def uint_get_dummies_default2pandas_workaround(df): # timeit(number=10) -> 6.99s
    """This function uses only modin, but defaults to pandas for every operation in this workaround."""
    uint8_dummies = pd.get_dummies(df)
    # Modin's astype will trigger the data import to OmniSci as well, so defaulting to pandas
    valid_dummies = uint8_dummies._default_to_pandas(pandas.DataFrame.astype, "int8")
    # Workaround for Modin issue #3370
    valid_dummies.columns = valid_dummies.columns.astype("str")
    return valid_dummies

ser = pd.Series(range(4096))

t1 = timeit.timeit(lambda: uint_get_dummies_only_pandas_workaround(ser), number=10)
t2 = timeit.timeit(lambda: uint_get_dummies_default2pandas_workaround(ser), number=10)

print(f"use only pandas: {t1}")
print(f"use modin, but fallback to pandas: {t2}")

```

</details>
> that's right. @gshimansky How does benchmark source code changes affect performance?

I cannot compare performance with and without the patch. Without the patch benchmark doesn't complete because of an exception.
It didn’t work earlier at all?
I remember that it worked but we don't know which commit broke it.

Commit message: Fix flake formatting

Signed-off-by: Dmitry Chigarev <dmitry.chigarev@intel.com>

Patch to review:
diff --git a/docs/release_notes/release_notes-0.14.0.rst b/docs/release_notes/release_notes-0.14.0.rst
index a2f1ba12594..a777f37cc67 100644
--- a/docs/release_notes/release_notes-0.14.0.rst
+++ b/docs/release_notes/release_notes-0.14.0.rst
@@ -13,6 +13,7 @@ Key Features and Updates
   * FIX-#4158: Do not print OmniSci logs to stdout by default (#4159)
   * FIX-#4177: Support read_feather from pathlike objects (#4177)
   * FIX-#4234: Upgrade pandas to 1.4.1 (#4235)
+  * FIX-#3368: support unsigned integers in OmniSci backend (#4256)
   * FIX-#4057: Allow reading an empty parquet file (#4075)  
   * FIX-#3884: Fix read_excel() dropping empty rows (#4161)
   * FIX-#4257: Fix Categorical() for scalar categories (#4258) 
diff --git a/modin/experimental/core/execution/native/implementations/omnisci_on_native/omnisci_worker.py b/modin/experimental/core/execution/native/implementations/omnisci_on_native/omnisci_worker.py
index 3c904551b9f..4700a898e17 100644
--- a/modin/experimental/core/execution/native/implementations/omnisci_on_native/omnisci_worker.py
+++ b/modin/experimental/core/execution/native/implementations/omnisci_on_native/omnisci_worker.py
@@ -22,6 +22,7 @@
 from .utils import PyDbEngine
 
 from modin.config import OmnisciFragmentSize, OmnisciLaunchParameters
+from modin.error_message import ErrorMessage
 
 
 class OmnisciServer:
@@ -139,7 +140,14 @@ def cast_to_compatible_types(table):
         schema = table.schema
         new_schema = schema
         need_cast = False
+        uint_to_int_cast = False
         new_cols = {}
+        uint_to_int_map = {
+            pa.uint8(): pa.int16(),
+            pa.uint16(): pa.int32(),
+            pa.uint32(): pa.int64(),
+            pa.uint64(): pa.int64(),  # May cause overflow
+        }
         for i, field in enumerate(schema):
             # Currently OmniSci doesn't support Arrow table import with
             # dictionary columns. Here we cast dictionaries until support
@@ -169,12 +177,35 @@ def cast_to_compatible_types(table):
                 )
                 new_schema = new_schema.set(i, new_field)
                 need_cast = True
+            # OmniSci doesn't support unsigned types
+            elif pa.types.is_unsigned_integer(field.type):
+                new_field = pa.field(
+                    field.name,
+                    uint_to_int_map[field.type],
+                    field.nullable,
+                    field.metadata,
+                )
+                new_schema = new_schema.set(i, new_field)
+                need_cast = True
+                uint_to_int_cast = True
+
+        # Such cast may affect the data, so we have to raise a warning about it
+        if uint_to_int_cast:
+            ErrorMessage.single_warning(
+                "OmniSci does not support unsigned integer types, such types will be rounded up to the signed equivalent."
+            )
 
         for i, col in new_cols.items():
             table = table.set_column(i, new_schema[i], col)
 
         if need_cast:
-            table = table.cast(new_schema)
+            try:
+                table = table.cast(new_schema)
+            except pa.lib.ArrowInvalid as e:
+                raise (OverflowError if uint_to_int_cast else RuntimeError)(
+                    "An error occurred when trying to convert unsupported by OmniSci 'dtypes' "
+                    + f"to the supported ones, the schema to cast was: \n{new_schema}."
+                ) from e
 
         return table
 
diff --git a/modin/experimental/core/execution/native/implementations/omnisci_on_native/test/test_dataframe.py b/modin/experimental/core/execution/native/implementations/omnisci_on_native/test/test_dataframe.py
index 9bf150bca1d..f59ad534f4e 100644
--- a/modin/experimental/core/execution/native/implementations/omnisci_on_native/test/test_dataframe.py
+++ b/modin/experimental/core/execution/native/implementations/omnisci_on_native/test/test_dataframe.py
@@ -37,6 +37,7 @@
     df_equals_with_non_stable_indices,
 )
 from modin.utils import try_cast_to_pandas
+from modin.pandas.utils import from_arrow
 
 from modin.experimental.core.execution.native.implementations.omnisci_on_native.partitioning.partition_manager import (
     OmnisciOnNativeDataframePartitionManager,
@@ -1932,6 +1933,80 @@ def fillna(df, **kwargs):
 
         run_and_compare(fillna, data=self.ok_data, force_lazy=False)
 
+    @pytest.mark.parametrize(
+        "md_df_constructor",
+        [
+            pytest.param(pd.DataFrame, id="from_pandas_dataframe"),
+            pytest.param(
+                lambda pd_df: from_arrow(pyarrow.Table.from_pandas(pd_df)),
+                id="from_pyarrow_table",
+            ),
+        ],
+    )
+    def test_uint(self, md_df_constructor):
+        """
+        Verify that unsigned integer data could be imported-exported via OmniSci with no errors.
+
+        Originally, OmniSci does not support unsigned integers, there's a logic in Modin that
+        upcasts unsigned types to the compatible ones prior importing to OmniSci.
+        """
+        pd_df = pandas.DataFrame(
+            {
+                "uint8_in_int_bounds": np.array([1, 2, 3], dtype="uint8"),
+                "uint8_out-of_int_bounds": np.array(
+                    [(2**8) - 1, (2**8) - 2, (2**8) - 3], dtype="uint8"
+                ),
+                "uint16_in_int_bounds": np.array([1, 2, 3], dtype="uint16"),
+                "uint16_out-of_int_bounds": np.array(
+                    [(2**16) - 1, (2**16) - 2, (2**16) - 3], dtype="uint16"
+                ),
+                "uint32_in_int_bounds": np.array([1, 2, 3], dtype="uint32"),
+                "uint32_out-of_int_bounds": np.array(
+                    [(2**32) - 1, (2**32) - 2, (2**32) - 3], dtype="uint32"
+                ),
+                "uint64_in_int_bounds": np.array([1, 2, 3], dtype="uint64"),
+            }
+        )
+        md_df = md_df_constructor(pd_df)
+
+        with ForceOmnisciImport(md_df) as instance:
+            md_df_exported = instance.export_frames()[0]
+            result = md_df_exported.values
+            reference = pd_df.values
+            np.testing.assert_array_equal(result, reference)
+
+    @pytest.mark.parametrize(
+        "md_df_constructor",
+        [
+            pytest.param(pd.DataFrame, id="from_pandas_dataframe"),
+            pytest.param(
+                lambda pd_df: from_arrow(pyarrow.Table.from_pandas(pd_df)),
+                id="from_pyarrow_table",
+            ),
+        ],
+    )
+    def test_uint_overflow(self, md_df_constructor):
+        """
+        Verify that the exception is arisen when overflow occurs due to 'uint -> int' compatibility conversion.
+
+        Originally, OmniSci does not support unsigned integers, there's a logic in Modin that upcasts
+        unsigned types to the compatible ones prior importing to OmniSci. This test ensures that the
+        error is arisen when such conversion causes a data loss.
+        """
+        md_df = md_df_constructor(
+            pandas.DataFrame(
+                {
+                    "col": np.array(
+                        [(2**64) - 1, (2**64) - 2, (2**64) - 3], dtype="uint64"
+                    )
+                }
+            )
+        )
+
+        with pytest.raises(OverflowError):
+            with ForceOmnisciImport(md_df):
+                pass
+
 
 class TestDropna:
     data = {
diff --git a/modin/experimental/core/execution/native/implementations/omnisci_on_native/test/utils.py b/modin/experimental/core/execution/native/implementations/omnisci_on_native/test/utils.py
index b23963f6f16..25233149508 100644
--- a/modin/experimental/core/execution/native/implementations/omnisci_on_native/test/utils.py
+++ b/modin/experimental/core/execution/native/implementations/omnisci_on_native/test/utils.py
@@ -25,6 +25,7 @@
     io_ops_bad_exc,
     eval_io as general_eval_io,
 )
+from ..df_algebra import FrameNode
 
 from modin.experimental.core.execution.native.implementations.omnisci_on_native.omnisci_worker import (
     OmnisciServer,
@@ -178,6 +179,32 @@ def __init__(self, *dfs):
     def __enter__(self):
         return self
 
+    def export_frames(self):
+        """
+        Export tables from OmniSci that was imported by this instance.
+
+        Returns
+        -------
+        list of modin DataFrames
+            A list of DataFrames whose payload is ``pyarrow.Table`` that was just exported from OmniSci.
+        """
+        result = []
+        for df, frame_id in self._imported_frames:
+            # Append `TransformNode`` selecting all the columns (SELECT * FROM frame_id)
+            df = df[df.columns.tolist()]
+            modin_frame = df._query_compiler._modin_frame
+            # Forcibly executing plan via OmniSci
+            new_partitions = modin_frame._partition_mgr_cls.run_exec_plan(
+                modin_frame._op,
+                modin_frame._index_cols,
+                modin_frame._dtypes,
+                modin_frame._table_cols,
+            )
+            modin_frame._partitions = new_partitions
+            modin_frame._op = FrameNode(modin_frame)
+            result.append(df)
+        return result
+
     def __exit__(self, exc_type, exc_val, exc_tb):
         for df, frame_id in self._imported_frames:
             actual_frame_id = df._query_compiler._modin_frame._partitions[0][0].frame_id


Code review comments:
diff_hunk: @@ -178,6 +179,32 @@ def __init__(self, *dfs):
     def __enter__(self):
         return self
 
+    def export_frames(self):
+        """
+        Export tables from OmniSci that was imported by this instance.
+
+        Returns
+        -------
+        list of modin DataFrames
+            A list of DataFrames whose payload is ``pyarrow.Table`` that was just exported from OmniSci.
+        """
+        result = []
+        for df, frame_id in self._imported_frames:
+            # Append `TransformNode`` selecting all the columns (SELECT * FROM frame_id)
+            df = df[df.columns.tolist()]
+            modin_frame = df._query_compiler._modin_frame
+            # Forcibly executing plan via OmniSci
path: modin/experimental/core/execution/native/implementations/omnisci_on_native/test/utils.py
text: @author:
```suggestion
            # Forcibly executing plan via OmniSci. We can't use `modin_frame._execute()` here
            # as it has a chance of running via pyarrow bypassing OmniSci
```