Repository: modin-project/modin
Pull request: FEAT-#5423: Add a NumPy API to Modin
<!--
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.
- [ ] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [ ] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [ ] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [ ] Resolves #5423 <!-- issue must be created for each patch -->
- [ ] tests added and passing
- [ ] 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 -->


Problem statement:
Add a NumPy API Layer to Modin
**Is your feature request related to a problem? Please describe.**
A clear and concise description of what the problem is. What kind of performance improvements would you like to see with this new API?
It would be nice to be able to use the NumPy API to manipulate Modin objects, i.e. have a `modin.numpy` package.

Hints from issue discussion:

Commit message: Add tests to CI

Signed-off-by: Rehan Durrani <rehan@ponder.io>

Patch to review:
diff --git a/.github/workflows/push-to-master.yml b/.github/workflows/push-to-master.yml
index 73a66367e05..3886886b596 100644
--- a/.github/workflows/push-to-master.yml
+++ b/.github/workflows/push-to-master.yml
@@ -55,6 +55,7 @@ jobs:
           python -m pytest modin/pandas/test/dataframe/test_udf.py
           python -m pytest modin/pandas/test/dataframe/test_window.py
           python -m pytest modin/pandas/test/test_series.py
+          python -m pytest modin/numpy/test/test_array.py
           python -m pytest modin/pandas/test/test_rolling.py
           python -m pytest modin/pandas/test/test_concat.py
           python -m pytest modin/pandas/test/test_groupby.py
@@ -121,6 +122,7 @@ jobs:
           - modin/pandas/test/dataframe/test_window.py
           - modin/pandas/test/dataframe/test_pickle.py
           - modin/pandas/test/test_series.py
+          - modin/numpy/test/test_array.py
           - modin/pandas/test/test_rolling.py
           - modin/pandas/test/test_concat.py
           - modin/pandas/test/test_groupby.py
diff --git a/.github/workflows/push.yml b/.github/workflows/push.yml
index 9e3f36c1ab2..bc01666948d 100644
--- a/.github/workflows/push.yml
+++ b/.github/workflows/push.yml
@@ -297,6 +297,7 @@ jobs:
       - run: python -m pytest -n 2 modin/pandas/test/dataframe/test_window.py
       - run: python -m pytest -n 2 modin/pandas/test/dataframe/test_pickle.py
       - run: python -m pytest -n 2 modin/pandas/test/test_series.py
+      - run: python -m pytest -n 2 modin/numpy/test/test_array.py
       - run: python -m pytest -n 2 modin/pandas/test/test_rolling.py
       - run: python -m pytest -n 2 modin/pandas/test/test_concat.py
         if: matrix.engine == 'python'
@@ -334,6 +335,7 @@ jobs:
           - modin/pandas/test/dataframe/test_window.py
           - modin/pandas/test/dataframe/test_pickle.py
           - modin/pandas/test/test_series.py
+          - modin/numpy/test/test_array.py
           - modin/pandas/test/test_rolling.py
           - modin/pandas/test/test_concat.py
           - modin/pandas/test/test_groupby.py
diff --git a/modin/config/envvars.py b/modin/config/envvars.py
index d16a463d34f..28c23214e1a 100644
--- a/modin/config/envvars.py
+++ b/modin/config/envvars.py
@@ -627,6 +627,11 @@ class TestReadFromPostgres(EnvironmentVariable, type=bool):
     varname = "MODIN_TEST_READ_FROM_POSTGRES"
     default = False
 
+class ExperimentalNumPyAPI(EnvironmentVariable, type=bool):
+    """Set to true to use Modin's experimental NumPy API."""
+
+    varname = "MODIN_EXPERIMENTAL_NUMPY_API"
+    default = False
 
 class ReadSqlEngine(EnvironmentVariable, type=str):
     """Engine to run `read_sql`."""
diff --git a/modin/numpy/__init__.py b/modin/numpy/__init__.py
new file mode 100644
index 00000000000..903ae6b9948
--- /dev/null
+++ b/modin/numpy/__init__.py
@@ -0,0 +1,16 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+from .arr import *
+from .math import *
+from .constants import *
diff --git a/modin/numpy/arr.py b/modin/numpy/arr.py
new file mode 100644
index 00000000000..f5b6ffa34e7
--- /dev/null
+++ b/modin/numpy/arr.py
@@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __floordiv__ = _floor_divide
+
+    def _power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.pow(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __pow__ = _power
+
+    def _prod(self, axis=None, out=None, keepdims=None, where=None):
+        if axis is None:
+            result = self._query_compiler.prod(axis=0).prod(axis=1)
+            return array(_query_compiler=result)
+        else:
+            result = self._query_compiler.prod(axis=axis)
+            return array(_query_compiler=result)
+
+    def _multiply(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.mul(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mul__ = _multiply
+
+    def _remainder(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.mod(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.NaN, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mod__ = _remainder
+
+    def _subtract(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.sub(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rsub(0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __sub__ = _subtract
+
+    def _sum(
+        self, axis=None, dtype=None, out=None, keepdims=None, initial=None, where=None
+    ):
+        result = self._query_compiler.sum(axis=axis)
+        if dtype is not None:
+            result = result.astype(dtype)
+        if out is not None:
+            out._query_compiler = result
+            return
+        if axis is None:
+            return
+        else:
+            new_ndim = self._ndim - 1
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    def _get_shape(self):
+        if self._ndim == 1:
+            return (len(self._query_compiler.index),)
+        return (len(self._query_compiler.index), len(self._query_compiler.columns))
+
+    def _set_shape(self, new_shape):
+        if not (isinstance(new_shape, int)) and not isinstance(new_shape, tuple):
+            raise TypeError(
+                f"expected a sequence of integers or a single integer, got '{new_shape}'"
+            )
+        elif isinstance(new_shape, tuple):
+            for dim in new_shape:
+                if not isinstance(dim, int):
+                    raise TypeError(
+                        f"'{type(dim)}' object cannot be interpreted as an integer"
+                    )
+        from math import prod
+
+        new_dimensions = new_shape if isinstance(new_shape, int) else prod(new_shape)
+        if new_dimensions != prod(self._get_shape()):
+            raise ValueError(
+                f"cannot reshape array of size {prod(self._get_shape)} into {new_shape if isinstance(new_shape, tuple) else (new_shape,)}"
+            )
+        if isinstance(new_shape, int):
+            qcs = []
+            for index_val in self._query_compiler.index[1:]:
+                qcs.append(
+                    self._query_compiler.getitem_row_array([index_val]).reset_index(
+                        drop=True
+                    )
+                )
+            self._query_compiler = (
+                self._query_compiler.getitem_row_array([self._query_compiler.index[0]])
+                .reset_index(drop=True)
+                .concat(1, qcs, ignore_index=True)
+            )
+            self._ndim = 1
+        else:
+            raise NotImplementedError(
+                "Reshaping from a 2D object to a 2D object is not currently supported!"
+            )
+
+    shape = property(_get_shape, _set_shape)
+
+    def __repr__(self):
+        return repr(self._to_numpy())
+
+    def _to_numpy(self):
+        arr = self._query_compiler.to_numpy()
+        if self._ndim == 1:
+            arr = arr.flatten()
+        return arr
diff --git a/modin/numpy/constants.py b/modin/numpy/constants.py
new file mode 100644
index 00000000000..0d7576516f9
--- /dev/null
+++ b/modin/numpy/constants.py
@@ -0,0 +1,31 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+# flake8: noqa
+from numpy import (
+    Inf,
+    Infinity,
+    NAN,
+    NINF,
+    NZERO,
+    NaN,
+    PINF,
+    PZERO,
+    e,
+    euler_gamma,
+    inf,
+    infty,
+    nan,
+    newaxis,
+    pi,
+)
diff --git a/modin/numpy/math.py b/modin/numpy/math.py
new file mode 100644
index 00000000000..bf6061eca80
--- /dev/null
+++ b/modin/numpy/math.py
@@ -0,0 +1,276 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+
+def absolute(
+    x, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x, "_absolute"):
+        return x._absolute(
+            out=out, where=where, casting=casting, order=order, dtype=dtype, subok=subok
+        )
+    return numpy.absolute(
+        x, out=out, where=where, casting=casting, order=order, dtype=dtype, subok=subok
+    )
+
+
+abs = absolute
+
+
+def add(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_add"):
+        return x1._add(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.add(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def all(a, axis=None, out=None, keepdims=None, where=None):
+    if hasattr(a, "_all"):
+        return a._all(axis=axis, out=out, keepdims=keepdims, where=where)
+    return numpy.all(a, axis=axis, out=out, keepdims=keepdims, where=where)
+
+
+def divide(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_divide"):
+        return x1._divide(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.divide(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def float_power(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_float_power"):
+        return x1._float_power(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.float_power(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def floor_divide(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_floor_divide"):
+        return x1._floor_divide(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.floor_divide(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def power(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_power"):
+        return x1._power(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.power(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def prod(a, axis=None, out=None, keepdims=None, where=None):
+    if hasattr(a, "_prod"):
+        return a._prod(axis=axis, out=out, keepdims=keepdims, where=where)
+    return numpy.prod(a, axis=axis, out=out, keepdims=keepdims, where=where)
+
+
+def multiply(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_multiply"):
+        return x1._multiply(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.multiply(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def remainder(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_remainder"):
+        return x1._remainder(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.remainder(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+mod = remainder
+
+
+def subtract(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_subtract"):
+        return x1._subtract(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.subtract(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def sum(arr, axis=None, dtype=None, out=None, keepdims=None, initial=None, where=None):
+    if hasattr(arr, "_sum"):
+        return arr._sum(axis)
+    else:
+        return numpy.sum(arr)
+
+
+def true_divide(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_divide"):
+        return x1._divide(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.divide(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
diff --git a/modin/numpy/test/__init__.py b/modin/numpy/test/__init__.py
new file mode 100644
index 00000000000..cae6413e559
--- /dev/null
+++ b/modin/numpy/test/__init__.py
@@ -0,0 +1,12 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
diff --git a/modin/numpy/test/test_array.py b/modin/numpy/test/test_array.py
new file mode 100644
index 00000000000..ef96b89c2fb
--- /dev/null
+++ b/modin/numpy/test/test_array.py
@@ -0,0 +1,60 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+import pytest
+import modin.numpy as np
+
+@pytest.mark.parametrize("size", [100, (2, 100), (100, 2), (1, 100), (100, 1)])
+def test_repr(size):
+    numpy_arr = numpy.random.randint(-100, 100, size=size)
+    modin_arr = np.array(numpy_arr)
+    assert repr(modin_arr) == repr(numpy_arr)
+
+@pytest.mark.parametrize("size", [100, (2, 100), (100, 2), (1, 100), (100, 1)])
+def test_shape(size):
+    numpy_arr = numpy.random.randint(-100, 100, size=size)
+    modin_arr = np.array(numpy_arr)
+    assert modin_arr.shape == numpy_arr.shape
+
+@pytest.mark.parametrize("operand1shape", [100, (3, 100)])
+@pytest.mark.parametrize("operand2shape", [100, (3, 100)])
+@pytest.mark.parametrize("operator", ["__add__", "__sub__", "__truediv__", "__mul__"])
+def test_basic_arithmetic_with_broadcast(operand1shape, operand2shape, operator):
+    """Test of operators that support broadcasting."""
+    operand1 = numpy.random.randint(-100, 100, size=operand1shape)
+    operand2 = numpy.random.randint(-100, 100, size=operand2shape)
+    modin_result = np.array(operand1).__getattribute__(operator)(np.array(operand2))
+    numpy_result = operand1.__getattribute__(operator)(operand2)
+    if operator != "__truediv__":
+        numpy.testing.assert_array_equal(modin_result._to_numpy(), numpy_result, err_msg=f"Binary Op {operator} failed.")
+    else:
+        # Truediv can have precision issues.
+        numpy.testing.assert_array_almost_equal(modin_result._to_numpy(), numpy_result, decimal=12, err_msg="Binary Op __truediv__ failed.")
+
+@pytest.mark.parametrize("operator", ["__pow__", "__floordiv__", "__mod__"])
+def test_complex_arithmetic(operator):
+    """Test of operators that do not yet support broadcasting"""
+    operand1 = numpy.random.randint(-100, 100, size=100)
+    lower_bound = -100 if operator != "__pow__" else 0
+    operand2 = numpy.random.randint(lower_bound, 100, size=100)
+    modin_result = np.array(operand1).__getattribute__(operator)(np.array(operand2))
+    numpy_result = operand1.__getattribute__(operator)(operand2)
+    numpy.testing.assert_array_almost_equal(modin_result._to_numpy(), numpy_result, decimal=12, err_msg=f"Binary Op {operator} failed on 1D arrays.")
+
+    operand1 = numpy.random.randint(-100, 100, size=(10, 10))
+    lower_bound = -100 if operator != "__pow__" else 0
+    operand2 = numpy.random.randint(lower_bound, 100, size=(10, 10))
+    modin_result = np.array(operand1).__getattribute__(operator)(np.array(operand2))
+    numpy_result = operand1.__getattribute__(operator)(operand2)
+    numpy.testing.assert_array_almost_equal(modin_result._to_numpy(), numpy_result, decimal=12, err_msg=f"Binary Op {operator} failed on 2D arrays.")
diff --git a/modin/pandas/base.py b/modin/pandas/base.py
index 19bdf24112c..9d48125af70 100644
--- a/modin/pandas/base.py
+++ b/modin/pandas/base.py
@@ -3242,6 +3242,12 @@ def to_numpy(
         """
         Convert the `BasePandasDataset` to a NumPy array.
         """
+        from modin.config import ExperimentalNumPyAPI
+        if ExperimentalNumPyAPI.get():
+            from ..numpy.arr import array
+
+            return array(_query_compiler=self._query_compiler, _ndim=2)
+        
         return self._query_compiler.to_numpy(
             dtype=dtype,
             copy=copy,
diff --git a/modin/pandas/series.py b/modin/pandas/series.py
index 81f87ec6b70..b0022b1f0b0 100644
--- a/modin/pandas/series.py
+++ b/modin/pandas/series.py
@@ -1948,15 +1948,21 @@ def to_numpy(
         """
         Return the NumPy ndarray representing the values in this Series or Index.
         """
-        return (
-            super(Series, self)
-            .to_numpy(
-                dtype=dtype,
-                copy=copy,
-                na_value=na_value,
+        from modin.config import ExperimentalNumPyAPI
+        if not ExperimentalNumPyAPI.get():
+            return (
+                super(Series, self)
+                .to_numpy(
+                    dtype=dtype,
+                    copy=copy,
+                    na_value=na_value,
+                )
+                .flatten()
             )
-            .flatten()
-        )
+        else:
+            from ..numpy.arr import array
+
+            return array(_query_compiler=self._query_compiler, _ndim=1)
 
     tolist = to_list
 


Code review comments:
diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
path: modin/numpy/arr.py
text: @user1:
```suggestion
            target_kwargs = {
                "dtype": None,
                "copy": True,
                "order": "K",
                "subok": False,
                "ndmin": 0,
                "like": None,
            }
            for key, value in target_kwargs.copy().items():
                if value == locals()[key]:
                    target_kwargs.pop(key)
                else:
                    target_kwargs[key] = locals()[key]
            arr = numpy.array(object, **target_kwargs)
```
also, why cannot we just pass those kwargs without dropping their default values?..

@author:
If we specify the kwargs with default types, we get an error:
```python
In [2]: import numpy as np

In [3]: np.array([1, 2], like=None)
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
Input In [3], in <cell line: 1>()
----> 1 np.array([1, 2], like=None)

TypeError: The `like` argument must be an array-like that implements the `__array_function__` protocol.
```

@user2:
I did some digging into the numpy source, and it seems that the value numpy uses for optional arguments is `numpy._NoValue` rather than `None` (as [here](https://github.com/numpy/numpy/blob/8cec82012694571156e8d7696307c848a7603b4e/numpy/core/fromnumeric.py#L1250))

@author:
Replacing `None` with numpy._NoValue and passing it in still causes the code to error out unfortunately.

@user1:
I would still better try to figure proper defaults... since they should be what we should take as function inputs at the frontend layer.

@author:
Even with the proper defaults that numpy uses, we still error out on passing them through - I think filtering the arguments and only passing the ones that aren't default makes sense to me, although I'm open to suggestions!

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
path: modin/numpy/arr.py
text: @user1:
why cannot we just `def __abs__` here?

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
path: modin/numpy/arr.py
text: @user1:
```suggestion
            caller, callee = (self, other) if self._ndim == 2 else (other, self)
```

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
path: modin/numpy/arr.py
text: @user1:
why cannot we `def __add__` here?

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
path: modin/numpy/arr.py
text: @user1:
## Explicit returns mixed with implicit (fall through) returns

Mixing implicit and explicit returns may indicate an error as implicit returns always return None.

[Show more details](https://github.com/modin-project/modin/security/code-scanning/499)

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
path: modin/numpy/arr.py
text: @user1:
why cannot we import that at the top of the module?

@author:
Moved all imports to the top!

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
path: modin/numpy/arr.py
text: @user1:
```suggestion
                if self.shape[0] == other.shape[0]:
                    matched_dimension = 0
                elif self.shape[1] == other.shape[1]:
                    matched_dimension = 1
                else:
                    raise ValueError(
                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
                    )

                if (
                    self.shape[matched_dimension ^ 1] == 1
                    or other.shape[matched_dimension ^ 1] == 1
                ):
                    # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
                    # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
                    return (self, other, self._ndim, {"broadcast": True, "axis": 1})
```

@user2:
Not sure if we wanted to leave this case for later, but this code doesn't work when one of the dimensions is 1, e.g. `np.array([[1] * 5] * 4) * np.array([2])` (a 5x4 array multiplied by a 1x1 array). Per numpy docs:
> When operating on two arrays, NumPy compares their shapes element-wise. It starts with the trailing (i.e. rightmost) dimension and works its way left. Two dimensions are compatible when
> 1. they are equal, or
> 2. one of them is 1.

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
path: modin/numpy/arr.py
text: @user1:
shouldn't this be `raise NotImplementedError`?

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
path: modin/numpy/arr.py
text: @user1:
shouldn't this be `raise NotImplementedError`?

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
path: modin/numpy/arr.py
text: @user1:
could you pls explain _why_?

and why `any(foo.to_pandas())` instead of `foo.any()`?

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
path: modin/numpy/arr.py
text: @user1:
what is `kwargs` and why no real args like `where` are passed through? 😉 

same in lots of other places

@author:
We're ignoring the more complicated kwargs from numpy for now, but the kwargs that binary_ops gives us helps the query compiler understand how to broadcast.

@user1:
The thing is, `kwargs` is not defined anywhere, and real `_add` args are simply ignored. Feels wrong.

@author:
This is in the old code - the new code handles the actual __add__ kwargs.

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __floordiv__ = _floor_divide
+
+    def _power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.pow(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __pow__ = _power
+
+    def _prod(self, axis=None, out=None, keepdims=None, where=None):
+        if axis is None:
+            result = self._query_compiler.prod(axis=0).prod(axis=1)
+            return array(_query_compiler=result)
+        else:
+            result = self._query_compiler.prod(axis=axis)
+            return array(_query_compiler=result)
+
+    def _multiply(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.mul(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mul__ = _multiply
+
+    def _remainder(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
path: modin/numpy/arr.py
text: @user1:
shouldn't this be `raise NotImplementedError`?

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
path: modin/numpy/arr.py
text: @user1:
```suggestion
            self._query_compiler = pd.Series(object)._query_compiler
```
don't see a point for intermediate variable

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __floordiv__ = _floor_divide
+
+    def _power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
path: modin/numpy/arr.py
text: @user1:
shouldn't this be `raise NotImplementedError`?

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __floordiv__ = _floor_divide
+
+    def _power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.pow(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __pow__ = _power
+
+    def _prod(self, axis=None, out=None, keepdims=None, where=None):
+        if axis is None:
+            result = self._query_compiler.prod(axis=0).prod(axis=1)
+            return array(_query_compiler=result)
+        else:
+            result = self._query_compiler.prod(axis=axis)
+            return array(_query_compiler=result)
+
+    def _multiply(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.mul(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mul__ = _multiply
+
+    def _remainder(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.mod(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.NaN, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mod__ = _remainder
+
+    def _subtract(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.sub(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rsub(0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __sub__ = _subtract
+
+    def _sum(
+        self, axis=None, dtype=None, out=None, keepdims=None, initial=None, where=None
+    ):
+        result = self._query_compiler.sum(axis=axis)
+        if dtype is not None:
+            result = result.astype(dtype)
+        if out is not None:
+            out._query_compiler = result
+            return
+        if axis is None:
+            return
path: modin/numpy/arr.py
text: @user1:
this feels weird, so it's not doing an in-place add but returns nothing?.. looks like a bug

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __floordiv__ = _floor_divide
+
+    def _power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.pow(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __pow__ = _power
+
+    def _prod(self, axis=None, out=None, keepdims=None, where=None):
+        if axis is None:
+            result = self._query_compiler.prod(axis=0).prod(axis=1)
+            return array(_query_compiler=result)
+        else:
+            result = self._query_compiler.prod(axis=axis)
+            return array(_query_compiler=result)
+
+    def _multiply(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.mul(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mul__ = _multiply
+
+    def _remainder(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.mod(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.NaN, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mod__ = _remainder
+
+    def _subtract(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.sub(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rsub(0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __sub__ = _subtract
+
+    def _sum(
+        self, axis=None, dtype=None, out=None, keepdims=None, initial=None, where=None
+    ):
+        result = self._query_compiler.sum(axis=axis)
+        if dtype is not None:
+            result = result.astype(dtype)
+        if out is not None:
+            out._query_compiler = result
+            return
+        if axis is None:
+            return
+        else:
+            new_ndim = self._ndim - 1
path: modin/numpy/arr.py
text: @user1:
why dimensions are reduced upon summation?

@author:
Summation here is a reduction that reduces one or more axes.

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __floordiv__ = _floor_divide
+
+    def _power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.pow(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __pow__ = _power
+
+    def _prod(self, axis=None, out=None, keepdims=None, where=None):
+        if axis is None:
+            result = self._query_compiler.prod(axis=0).prod(axis=1)
+            return array(_query_compiler=result)
+        else:
+            result = self._query_compiler.prod(axis=axis)
+            return array(_query_compiler=result)
+
+    def _multiply(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.mul(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mul__ = _multiply
+
+    def _remainder(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.mod(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.NaN, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mod__ = _remainder
+
+    def _subtract(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.sub(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rsub(0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __sub__ = _subtract
+
+    def _sum(
+        self, axis=None, dtype=None, out=None, keepdims=None, initial=None, where=None
+    ):
+        result = self._query_compiler.sum(axis=axis)
+        if dtype is not None:
+            result = result.astype(dtype)
+        if out is not None:
+            out._query_compiler = result
+            return
+        if axis is None:
+            return
+        else:
+            new_ndim = self._ndim - 1
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    def _get_shape(self):
+        if self._ndim == 1:
+            return (len(self._query_compiler.index),)
+        return (len(self._query_compiler.index), len(self._query_compiler.columns))
path: modin/numpy/arr.py
text: @user1:
shouldn't this be an `assert self._ndim == 2` here?

@user2:
This also needs a case for `_ndim == 0`:
```
>>> np.array(1).ndim  # passing a scalar into the constructor
0
```

@author:
For now, we're only focusing on 1D and 2D cases @user2. @user1 why would we need this assert?

@user1:
To immediately know something is not behaving the way we expected. That scalar is an awesome example - a user can feed a scalar and receive some cryptic error message instead of somewhat clear assertion (with a pinpointed place of erroring out to boot).

@author:
Makes sense to me - I'll go ahead and add it!

@user2:
Did you put this assert in? I don't see it in this version of the code.

@author:
I added the assert in the __init__. Does that work?

@user2:
sure, that's fine

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __floordiv__ = _floor_divide
+
+    def _power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.pow(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __pow__ = _power
+
+    def _prod(self, axis=None, out=None, keepdims=None, where=None):
+        if axis is None:
+            result = self._query_compiler.prod(axis=0).prod(axis=1)
+            return array(_query_compiler=result)
+        else:
+            result = self._query_compiler.prod(axis=axis)
+            return array(_query_compiler=result)
+
+    def _multiply(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.mul(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mul__ = _multiply
+
+    def _remainder(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.mod(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.NaN, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mod__ = _remainder
+
+    def _subtract(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.sub(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rsub(0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __sub__ = _subtract
+
+    def _sum(
+        self, axis=None, dtype=None, out=None, keepdims=None, initial=None, where=None
+    ):
+        result = self._query_compiler.sum(axis=axis)
+        if dtype is not None:
+            result = result.astype(dtype)
+        if out is not None:
+            out._query_compiler = result
+            return
+        if axis is None:
+            return
+        else:
+            new_ndim = self._ndim - 1
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    def _get_shape(self):
+        if self._ndim == 1:
+            return (len(self._query_compiler.index),)
+        return (len(self._query_compiler.index), len(self._query_compiler.columns))
+
+    def _set_shape(self, new_shape):
+        if not (isinstance(new_shape, int)) and not isinstance(new_shape, tuple):
+            raise TypeError(
+                f"expected a sequence of integers or a single integer, got '{new_shape}'"
+            )
+        elif isinstance(new_shape, tuple):
+            for dim in new_shape:
+                if not isinstance(dim, int):
+                    raise TypeError(
+                        f"'{type(dim)}' object cannot be interpreted as an integer"
+                    )
+        from math import prod
path: modin/numpy/arr.py
text: @user1:
I think we try to import things on top; to not confuse readers I would suggest doing `import math` on top and using `math.prod()` here

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __floordiv__ = _floor_divide
+
+    def _power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.pow(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __pow__ = _power
+
+    def _prod(self, axis=None, out=None, keepdims=None, where=None):
+        if axis is None:
+            result = self._query_compiler.prod(axis=0).prod(axis=1)
+            return array(_query_compiler=result)
+        else:
+            result = self._query_compiler.prod(axis=axis)
+            return array(_query_compiler=result)
+
+    def _multiply(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.mul(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mul__ = _multiply
+
+    def _remainder(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.mod(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.NaN, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mod__ = _remainder
+
+    def _subtract(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.sub(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rsub(0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __sub__ = _subtract
+
+    def _sum(
+        self, axis=None, dtype=None, out=None, keepdims=None, initial=None, where=None
+    ):
+        result = self._query_compiler.sum(axis=axis)
+        if dtype is not None:
+            result = result.astype(dtype)
+        if out is not None:
+            out._query_compiler = result
+            return
+        if axis is None:
+            return
+        else:
+            new_ndim = self._ndim - 1
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    def _get_shape(self):
+        if self._ndim == 1:
+            return (len(self._query_compiler.index),)
+        return (len(self._query_compiler.index), len(self._query_compiler.columns))
+
+    def _set_shape(self, new_shape):
+        if not (isinstance(new_shape, int)) and not isinstance(new_shape, tuple):
+            raise TypeError(
+                f"expected a sequence of integers or a single integer, got '{new_shape}'"
+            )
+        elif isinstance(new_shape, tuple):
+            for dim in new_shape:
+                if not isinstance(dim, int):
+                    raise TypeError(
+                        f"'{type(dim)}' object cannot be interpreted as an integer"
+                    )
+        from math import prod
+
+        new_dimensions = new_shape if isinstance(new_shape, int) else prod(new_shape)
+        if new_dimensions != prod(self._get_shape()):
+            raise ValueError(
+                f"cannot reshape array of size {prod(self._get_shape)} into {new_shape if isinstance(new_shape, tuple) else (new_shape,)}"
+            )
+        if isinstance(new_shape, int):
+            qcs = []
+            for index_val in self._query_compiler.index[1:]:
+                qcs.append(
+                    self._query_compiler.getitem_row_array([index_val]).reset_index(
+                        drop=True
+                    )
+                )
path: modin/numpy/arr.py
text: @user1:
```suggestion
            qcs = [
                self._query_compiler.getitem_row_array([index_val]).reset_index(
                      drop=True
                ) for index_val in self._query_compiler.index[1:]
            ]
```

diff_hunk: @@ -0,0 +1,276 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+
+def absolute(
+    x, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x, "_absolute"):
+        return x._absolute(
+            out=out, where=where, casting=casting, order=order, dtype=dtype, subok=subok
+        )
+    return numpy.absolute(
+        x, out=out, where=where, casting=casting, order=order, dtype=dtype, subok=subok
+    )
+
+
+abs = absolute
+
+
+def add(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_add"):
+        return x1._add(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.add(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def all(a, axis=None, out=None, keepdims=None, where=None):
+    if hasattr(a, "_all"):
+        return a._all(axis=axis, out=out, keepdims=keepdims, where=where)
+    return numpy.all(a, axis=axis, out=out, keepdims=keepdims, where=where)
+
+
+def divide(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_divide"):
+        return x1._divide(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.divide(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def float_power(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_float_power"):
+        return x1._float_power(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.float_power(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def floor_divide(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_floor_divide"):
+        return x1._floor_divide(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.floor_divide(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def power(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_power"):
+        return x1._power(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.power(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def prod(a, axis=None, out=None, keepdims=None, where=None):
+    if hasattr(a, "_prod"):
+        return a._prod(axis=axis, out=out, keepdims=keepdims, where=where)
+    return numpy.prod(a, axis=axis, out=out, keepdims=keepdims, where=where)
+
+
+def multiply(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_multiply"):
+        return x1._multiply(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.multiply(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def remainder(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_remainder"):
+        return x1._remainder(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.remainder(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+mod = remainder
+
+
+def subtract(
+    x1, x2, out=None, where=True, casting="same_kind", order="K", dtype=None, subok=True
+):
+    if hasattr(x1, "_subtract"):
+        return x1._subtract(
+            x2,
+            out=out,
+            where=where,
+            casting=casting,
+            order=order,
+            dtype=dtype,
+            subok=subok,
+        )
+    return numpy.subtract(
+        x1,
+        x2,
+        out=out,
+        where=where,
+        casting=casting,
+        order=order,
+        dtype=dtype,
+        subok=subok,
+    )
+
+
+def sum(arr, axis=None, dtype=None, out=None, keepdims=None, initial=None, where=None):
+    if hasattr(arr, "_sum"):
+        return arr._sum(axis)
+    else:
+        return numpy.sum(arr)
path: modin/numpy/math.py
text: @user1:
why other args are lost?

diff_hunk: @@ -0,0 +1,60 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+import pytest
+import modin.numpy as np
+
+@pytest.mark.parametrize("size", [100, (2, 100), (100, 2), (1, 100), (100, 1)])
+def test_repr(size):
+    numpy_arr = numpy.random.randint(-100, 100, size=size)
+    modin_arr = np.array(numpy_arr)
+    assert repr(modin_arr) == repr(numpy_arr)
+
+@pytest.mark.parametrize("size", [100, (2, 100), (100, 2), (1, 100), (100, 1)])
+def test_shape(size):
+    numpy_arr = numpy.random.randint(-100, 100, size=size)
+    modin_arr = np.array(numpy_arr)
+    assert modin_arr.shape == numpy_arr.shape
+
+@pytest.mark.parametrize("operand1shape", [100, (3, 100)])
+@pytest.mark.parametrize("operand2shape", [100, (3, 100)])
+@pytest.mark.parametrize("operator", ["__add__", "__sub__", "__truediv__", "__mul__"])
+def test_basic_arithmetic_with_broadcast(operand1shape, operand2shape, operator):
+    """Test of operators that support broadcasting."""
+    operand1 = numpy.random.randint(-100, 100, size=operand1shape)
+    operand2 = numpy.random.randint(-100, 100, size=operand2shape)
+    modin_result = np.array(operand1).__getattribute__(operator)(np.array(operand2))
+    numpy_result = operand1.__getattribute__(operator)(operand2)
path: modin/numpy/test/test_array.py
text: @user1:
```suggestion
    modin_result = getattr(np.array(operand1), operator)(np.array(operand2))
    numpy_result = getattr(operand1, operator)(operand2)
```
`getattr()` looks shorter

diff_hunk: @@ -0,0 +1,60 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+import pytest
+import modin.numpy as np
+
+@pytest.mark.parametrize("size", [100, (2, 100), (100, 2), (1, 100), (100, 1)])
+def test_repr(size):
+    numpy_arr = numpy.random.randint(-100, 100, size=size)
+    modin_arr = np.array(numpy_arr)
+    assert repr(modin_arr) == repr(numpy_arr)
+
+@pytest.mark.parametrize("size", [100, (2, 100), (100, 2), (1, 100), (100, 1)])
+def test_shape(size):
+    numpy_arr = numpy.random.randint(-100, 100, size=size)
+    modin_arr = np.array(numpy_arr)
+    assert modin_arr.shape == numpy_arr.shape
+
+@pytest.mark.parametrize("operand1shape", [100, (3, 100)])
+@pytest.mark.parametrize("operand2shape", [100, (3, 100)])
+@pytest.mark.parametrize("operator", ["__add__", "__sub__", "__truediv__", "__mul__"])
+def test_basic_arithmetic_with_broadcast(operand1shape, operand2shape, operator):
+    """Test of operators that support broadcasting."""
+    operand1 = numpy.random.randint(-100, 100, size=operand1shape)
+    operand2 = numpy.random.randint(-100, 100, size=operand2shape)
+    modin_result = np.array(operand1).__getattribute__(operator)(np.array(operand2))
+    numpy_result = operand1.__getattribute__(operator)(operand2)
+    if operator != "__truediv__":
+        numpy.testing.assert_array_equal(modin_result._to_numpy(), numpy_result, err_msg=f"Binary Op {operator} failed.")
+    else:
+        # Truediv can have precision issues.
+        numpy.testing.assert_array_almost_equal(modin_result._to_numpy(), numpy_result, decimal=12, err_msg="Binary Op __truediv__ failed.")
+
+@pytest.mark.parametrize("operator", ["__pow__", "__floordiv__", "__mod__"])
+def test_complex_arithmetic(operator):
+    """Test of operators that do not yet support broadcasting"""
+    operand1 = numpy.random.randint(-100, 100, size=100)
+    lower_bound = -100 if operator != "__pow__" else 0
+    operand2 = numpy.random.randint(lower_bound, 100, size=100)
+    modin_result = np.array(operand1).__getattribute__(operator)(np.array(operand2))
+    numpy_result = operand1.__getattribute__(operator)(operand2)
+    numpy.testing.assert_array_almost_equal(modin_result._to_numpy(), numpy_result, decimal=12, err_msg=f"Binary Op {operator} failed on 1D arrays.")
+
+    operand1 = numpy.random.randint(-100, 100, size=(10, 10))
+    lower_bound = -100 if operator != "__pow__" else 0
+    operand2 = numpy.random.randint(lower_bound, 100, size=(10, 10))
+    modin_result = np.array(operand1).__getattribute__(operator)(np.array(operand2))
+    numpy_result = operand1.__getattribute__(operator)(operand2)
+    numpy.testing.assert_array_almost_equal(modin_result._to_numpy(), numpy_result, decimal=12, err_msg=f"Binary Op {operator} failed on 2D arrays.")
path: modin/numpy/test/test_array.py
text: @user1:
```suggestion
    for size, textdim in ((100, "1D"), ((10, 10), "2D")):
        operand1 = numpy.random.randint(-100, 100, size=size)
        lower_bound = -100 if operator != "__pow__" else 0
        operand2 = numpy.random.randint(lower_bound, 100, size=size)
        modin_result = getattr(np.array(operand1), operator)(np.array(operand2))
        numpy_result = getattr(operand1, operator)(operand2)
        numpy.testing.assert_array_almost_equal(modin_result._to_numpy(), numpy_result, decimal=12, err_msg=f"Binary Op {operator} failed on {textdim} arrays.")
```

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
+        broadcast = self._ndim != other._ndim
+        if broadcast:
+            # In this case, we have a 1D object doing a binary op with a 2D object
+            caller = self if self._ndim == 2 else other
+            callee = other if self._ndim == 2 else self
+            return (caller, callee, caller._ndim, {"broadcast": broadcast, "axis": 1})
+        else:
+            if self.shape != other.shape:
+                # In this case, we either have two mismatched objects trying to do an operation
+                # or a nested 1D object that must be broadcasted trying to do an operation.
+                matched_dimension = None
+                if self.shape[0] == other.shape[0]:
+                    matched_dimension = 0
+                elif self.shape[1] == other.shape[1]:
+                    matched_dimension = 1
+                if matched_dimension is not None:
+                    if (
+                        self.shape[matched_dimension ^ 1] == 1
+                        or other.shape[matched_dimension ^ 1] == 1
+                    ):
+                        # caller = self if other.shape[matched_dimension ^ 1] == 1 else other
+                        # callee = other if other.shape[matched_dimension ^ 1] == 1 else self
+                        return (self, other, self._ndim, {"broadcast": True, "axis": 1})
+                else:
+                    raise ValueError(
+                        f"operands could not be broadcast together with shapes {self.shape} {other.shape}"
+                    )
+            else:
+                return (self, other, self._ndim, {"broadcast": False})
+
+    def _add(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.add(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __add__ = _add
+
+    def _divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.truediv(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rtruediv(1)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __truediv__ = _divide
+
+    def _float_power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        pass
+
+    def _floor_divide(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.floordiv(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.inf, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __floordiv__ = _floor_divide
+
+    def _power(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.pow(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __pow__ = _power
+
+    def _prod(self, axis=None, out=None, keepdims=None, where=None):
+        if axis is None:
+            result = self._query_compiler.prod(axis=0).prod(axis=1)
+            return array(_query_compiler=result)
+        else:
+            result = self._query_compiler.prod(axis=axis)
+            return array(_query_compiler=result)
+
+    def _multiply(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.mul(callee._query_compiler, **kwargs)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mul__ = _multiply
+
+    def _remainder(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        if caller != self:
+            # No workaround possible until broadcasting fixed. GH#5529.
+            pass
+        result = caller._query_compiler.mod(callee._query_compiler, **kwargs)
+        if any(callee._query_compiler.eq(0).to_pandas()):
+            result = result.replace(numpy.NaN, 0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __mod__ = _remainder
+
+    def _subtract(
+        self,
+        x2,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        caller, callee, new_ndim, kwargs = self._binary_op(x2)
+        result = caller._query_compiler.sub(callee._query_compiler, **kwargs)
+        if caller != self:
+            result = result.rsub(0)
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    __sub__ = _subtract
+
+    def _sum(
+        self, axis=None, dtype=None, out=None, keepdims=None, initial=None, where=None
+    ):
+        result = self._query_compiler.sum(axis=axis)
+        if dtype is not None:
+            result = result.astype(dtype)
+        if out is not None:
+            out._query_compiler = result
+            return
+        if axis is None:
+            return
+        else:
+            new_ndim = self._ndim - 1
+        return array(_query_compiler=result, _ndim=new_ndim)
+
+    def _get_shape(self):
+        if self._ndim == 1:
+            return (len(self._query_compiler.index),)
+        return (len(self._query_compiler.index), len(self._query_compiler.columns))
+
+    def _set_shape(self, new_shape):
+        if not (isinstance(new_shape, int)) and not isinstance(new_shape, tuple):
+            raise TypeError(
+                f"expected a sequence of integers or a single integer, got '{new_shape}'"
+            )
+        elif isinstance(new_shape, tuple):
+            for dim in new_shape:
+                if not isinstance(dim, int):
+                    raise TypeError(
+                        f"'{type(dim)}' object cannot be interpreted as an integer"
+                    )
+        from math import prod
+
+        new_dimensions = new_shape if isinstance(new_shape, int) else prod(new_shape)
+        if new_dimensions != prod(self._get_shape()):
+            raise ValueError(
+                f"cannot reshape array of size {prod(self._get_shape)} into {new_shape if isinstance(new_shape, tuple) else (new_shape,)}"
+            )
+        if isinstance(new_shape, int):
+            qcs = []
+            for index_val in self._query_compiler.index[1:]:
+                qcs.append(
+                    self._query_compiler.getitem_row_array([index_val]).reset_index(
+                        drop=True
+                    )
+                )
+            self._query_compiler = (
+                self._query_compiler.getitem_row_array([self._query_compiler.index[0]])
+                .reset_index(drop=True)
+                .concat(1, qcs, ignore_index=True)
+            )
+            self._ndim = 1
+        else:
+            raise NotImplementedError(
+                "Reshaping from a 2D object to a 2D object is not currently supported!"
+            )
+
+    shape = property(_get_shape, _set_shape)
+
+    def __repr__(self):
+        return repr(self._to_numpy())
+
+    def _to_numpy(self):
+        arr = self._query_compiler.to_numpy()
path: modin/numpy/arr.py
text: @user1:
also needs to cover the `_ndim == 0` case (`np.array(1)`)

diff_hunk: @@ -0,0 +1,333 @@
+# Licensed to Modin Development Team under one or more contributor license agreements.
+# See the NOTICE file distributed with this work for additional information regarding
+# copyright ownership.  The Modin Development Team licenses this file to you under the
+# Apache License, Version 2.0 (the "License"); you may not use this file except in
+# compliance with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
+# ANY KIND, either express or implied. See the License for the specific language
+# governing permissions and limitations under the License.
+
+import numpy
+
+from pandas.core.dtypes.common import is_list_like
+from modin.error_message import ErrorMessage
+
+
+class array(object):
+    def __init__(
+        self,
+        object=None,
+        dtype=None,
+        *,
+        copy=True,
+        order="K",
+        subok=False,
+        ndmin=0,
+        like=None,
+        _query_compiler=None,
+        _ndim=None,
+    ):
+        if _query_compiler is not None:
+            self._query_compiler = _query_compiler
+            self._ndim = _ndim
+        elif is_list_like(object) and not is_list_like(object[0]):
+            import modin.pandas as pd
+
+            qc = pd.Series(object)._query_compiler
+            self._query_compiler = qc
+            self._ndim = 1
+        else:
+            expected_kwargs = {
+                "dtype": None,
+                "copy": True,
+                "order": "K",
+                "subok": False,
+                "ndmin": 0,
+                "like": None,
+            }
+            rcvd_kwargs = {
+                "dtype": dtype,
+                "copy": copy,
+                "order": order,
+                "subok": subok,
+                "ndmin": ndmin,
+                "like": like,
+            }
+            for key, value in rcvd_kwargs.copy().items():
+                if value == expected_kwargs[key]:
+                    rcvd_kwargs.pop(key)
+            arr = numpy.array(object, **rcvd_kwargs)
+            self._ndim = len(arr.shape)
+            if self._ndim > 2:
+                ErrorMessage.not_implemented(
+                    "NumPy arrays with dimensions higher than 2 are not yet supported."
+                )
+            import modin.pandas as pd
+
+            self._query_compiler = pd.DataFrame(arr)._query_compiler
+
+    def _absolute(
+        self,
+        out=None,
+        where=True,
+        casting="same_kind",
+        order="K",
+        dtype=None,
+        subok=True,
+    ):
+        result = self._query_compiler.abs()
+        return array(_query_compiler=result, _ndim=self._ndim)
+
+    __abs__ = _absolute
+
+    def _binary_op(self, other):
path: modin/numpy/arr.py
text: @user1:
We also have to discard the original indices, otherwise, the binary op implementation will try to align indices as it still remembers them:
```python
import modin.config as cfg
import modin.pandas as pd

df1 = pd.DataFrame({"a": [1, 2], "b": [3, 4]})
df2 = pd.DataFrame({"b": [10, 10], "a": [100, 100]})

print("Regular array:")
print(df1.to_numpy() * df2.to_numpy())

cfg.ExperimentalNumPyAPI.put(True)
print("Experimental array:")
print(df1.to_numpy() * df2.to_numpy())
```
Output:
```
Regular array:
[[ 10 300]
 [ 20 400]]
Experimental array:
array([[100,  30],
       [200,  40]])
```
I would personally prefer the indices to be discarded by the `query_compiler.binary_op` implementation itself using some kind of `ignore_indices=True` flag passed from here. This way we could easily support a commonly used case `md_series[subset1] * md_series[subset2].values` in the future.

@user2:
Even if we discard the original indices, there will be certain cases that won't work.

```python
import pandas

df1 = pandas.DataFrame({"a": [1, 2], "b": [3, 4]})
df2 = pandas.DataFrame({"c": [1, 2]})
df1.columns = [0, 1]
df2.columns = [0]

df1 + df2
   0   1
0  2 NaN
1  4 NaN

df1.values + df2.values
array([[2, 4],
       [4, 6]])
```

We could convert pandas DataFrames to NumPy arrays in the compute kernels to get proper behavior for modin.numpy. However, I think that really mixes up pandas and numpy storage formats. So I would recommend to start implementing the Modin layers for modin.numpy to make the things work correctly and get the proper Modin structure without mixing up the things.

@user1:
so for now we want basic support and I propose that at least a case like this should work in this branch to proceed: cc @author 
```python
import modin.pandas as pd

df1 = pd.DataFrame({"a": [1, 2, 3, 4], "b": [3, 4, 5, 6], "c": [10, 20, 30, 40]})
df2 = pd.DataFrame({"x": [11, 12, 13, 14], "y": [15, 4, 5, 6], "z": [4, 3, 2, 1]})

indexer1 = pd.Series([1, 0, 2]).to_numpy()
indexer2 = pd.Series([3, 2, 1]).to_numpy()

# First and second '*' operands have completely different indices but '.to_numpy()' should ignore them
res = df1[["a", "c"]].to_numpy()[indexer1] * df2[["x", "y"]].to_numpy()[indexer2]
print(res)
```

@author:
Fixed!