We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Fix defensive assertions for Ray Client users
Related to https://github.com/ray-project/ray/issues/14042

For now, we will check for both types.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `modin/engines/ray/pandas_on_ray/frame/partition.py`
Content:
```
1 # Licensed to Modin Development Team under one or more contributor license agreements.
2 # See the NOTICE file distributed with this work for additional information regarding
3 # copyright ownership.  The Modin Development Team licenses this file to you under the
4 # Apache License, Version 2.0 (the "License"); you may not use this file except in
5 # compliance with the License.  You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software distributed under
10 # the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
11 # ANY KIND, either express or implied. See the License for the specific language
12 # governing permissions and limitations under the License.
13 
14 import pandas
15 
16 from modin.data_management.utils import length_fn_pandas, width_fn_pandas
17 from modin.engines.base.frame.partition import BaseFramePartition
18 from modin.engines.ray.utils import handle_ray_task_error
19 
20 import ray
21 from ray.worker import RayTaskError
22 from ray.services import get_node_ip_address
23 
24 
25 class PandasOnRayFramePartition(BaseFramePartition):
26     def __init__(self, object_id, length=None, width=None, ip=None, call_queue=None):
27         assert type(object_id) is ray.ObjectRef
28 
29         self.oid = object_id
30         if call_queue is None:
31             call_queue = []
32         self.call_queue = call_queue
33         self._length_cache = length
34         self._width_cache = width
35         self._ip_cache = ip
36 
37     def get(self):
38         """Gets the object out of the plasma store.
39 
40         Returns:
41             The object from the plasma store.
42         """
43         if len(self.call_queue):
44             self.drain_call_queue()
45         try:
46             return ray.get(self.oid)
47         except RayTaskError as e:
48             handle_ray_task_error(e)
49 
50     def apply(self, func, **kwargs):
51         """Apply a function to the object stored in this partition.
52 
53         Note: It does not matter if func is callable or an ObjectRef. Ray will
54             handle it correctly either way. The keyword arguments are sent as a
55             dictionary.
56 
57         Args:
58             func: The function to apply.
59 
60         Returns:
61             A RayRemotePartition object.
62         """
63         oid = self.oid
64         call_queue = self.call_queue + [(func, kwargs)]
65         result, length, width, ip = deploy_ray_func.remote(call_queue, oid)
66         return PandasOnRayFramePartition(result, length, width, ip)
67 
68     def add_to_apply_calls(self, func, **kwargs):
69         return PandasOnRayFramePartition(
70             self.oid, call_queue=self.call_queue + [(func, kwargs)]
71         )
72 
73     def drain_call_queue(self):
74         if len(self.call_queue) == 0:
75             return
76         oid = self.oid
77         call_queue = self.call_queue
78         (
79             self.oid,
80             self._length_cache,
81             self._width_cache,
82             self._ip_cache,
83         ) = deploy_ray_func.remote(call_queue, oid)
84         self.call_queue = []
85 
86     def wait(self):
87         self.drain_call_queue()
88         try:
89             ray.wait([self.oid])
90         except RayTaskError as e:
91             handle_ray_task_error(e)
92 
93     def __copy__(self):
94         return PandasOnRayFramePartition(
95             self.oid,
96             length=self._length_cache,
97             width=self._width_cache,
98             ip=self._ip_cache,
99             call_queue=self.call_queue,
100         )
101 
102     def to_pandas(self):
103         """Convert the object stored in this partition to a Pandas DataFrame.
104 
105         Returns:
106             A Pandas DataFrame.
107         """
108         dataframe = self.get()
109         assert type(dataframe) is pandas.DataFrame or type(dataframe) is pandas.Series
110         return dataframe
111 
112     def to_numpy(self, **kwargs):
113         """
114         Convert the object stored in this partition to a NumPy array.
115 
116         Returns
117         -------
118             A NumPy array.
119         """
120         return self.apply(lambda df, **kwargs: df.to_numpy(**kwargs)).get()
121 
122     def mask(self, row_indices, col_indices):
123         if (
124             (isinstance(row_indices, slice) and row_indices == slice(None))
125             or (
126                 not isinstance(row_indices, slice)
127                 and self._length_cache is not None
128                 and len(row_indices) == self._length_cache
129             )
130         ) and (
131             (isinstance(col_indices, slice) and col_indices == slice(None))
132             or (
133                 not isinstance(col_indices, slice)
134                 and self._width_cache is not None
135                 and len(col_indices) == self._width_cache
136             )
137         ):
138             return self.__copy__()
139 
140         new_obj = self.add_to_apply_calls(
141             lambda df: pandas.DataFrame(df.iloc[row_indices, col_indices])
142         )
143         if not isinstance(row_indices, slice):
144             new_obj._length_cache = len(row_indices)
145         if not isinstance(col_indices, slice):
146             new_obj._width_cache = len(col_indices)
147         return new_obj
148 
149     @classmethod
150     def put(cls, obj):
151         """Put an object in the Plasma store and wrap it in this object.
152 
153         Args:
154             obj: The object to be put.
155 
156         Returns:
157             A `RayRemotePartition` object.
158         """
159         return PandasOnRayFramePartition(ray.put(obj), len(obj.index), len(obj.columns))
160 
161     @classmethod
162     def preprocess_func(cls, func):
163         """Put a callable function into the plasma store for use in `apply`.
164 
165         Args:
166             func: The function to preprocess.
167 
168         Returns:
169             A ray.ObjectRef.
170         """
171         return ray.put(func)
172 
173     def length(self):
174         if self._length_cache is None:
175             if len(self.call_queue):
176                 self.drain_call_queue()
177             else:
178                 self._length_cache, self._width_cache = get_index_and_columns.remote(
179                     self.oid
180                 )
181         if isinstance(self._length_cache, ray.ObjectRef):
182             try:
183                 self._length_cache = ray.get(self._length_cache)
184             except RayTaskError as e:
185                 handle_ray_task_error(e)
186         return self._length_cache
187 
188     def width(self):
189         if self._width_cache is None:
190             if len(self.call_queue):
191                 self.drain_call_queue()
192             else:
193                 self._length_cache, self._width_cache = get_index_and_columns.remote(
194                     self.oid
195                 )
196         if isinstance(self._width_cache, ray.ObjectRef):
197             try:
198                 self._width_cache = ray.get(self._width_cache)
199             except RayTaskError as e:
200                 handle_ray_task_error(e)
201         return self._width_cache
202 
203     def ip(self):
204         if self._ip_cache is None:
205             if len(self.call_queue):
206                 self.drain_call_queue()
207             else:
208                 self._ip_cache = self.apply(lambda df: df)._ip_cache
209         if isinstance(self._ip_cache, ray.ObjectID):
210             try:
211                 self._ip_cache = ray.get(self._ip_cache)
212             except RayTaskError as e:
213                 handle_ray_task_error(e)
214         return self._ip_cache
215 
216     @classmethod
217     def length_extraction_fn(cls):
218         return length_fn_pandas
219 
220     @classmethod
221     def width_extraction_fn(cls):
222         return width_fn_pandas
223 
224     @classmethod
225     def empty(cls):
226         return cls.put(pandas.DataFrame())
227 
228 
229 @ray.remote(num_returns=2)
230 def get_index_and_columns(df):
231     return len(df.index), len(df.columns)
232 
233 
234 @ray.remote(num_returns=4)
235 def deploy_ray_func(call_queue, partition):  # pragma: no cover
236     def deserialize(obj):
237         if isinstance(obj, ray.ObjectRef):
238             return ray.get(obj)
239         return obj
240 
241     if len(call_queue) > 1:
242         for func, kwargs in call_queue[:-1]:
243             func = deserialize(func)
244             kwargs = deserialize(kwargs)
245             try:
246                 partition = func(partition, **kwargs)
247             except ValueError:
248                 partition = func(partition.copy(), **kwargs)
249     func, kwargs = call_queue[-1]
250     func = deserialize(func)
251     kwargs = deserialize(kwargs)
252     try:
253         result = func(partition, **kwargs)
254     # Sometimes Arrow forces us to make a copy of an object before we operate on it. We
255     # don't want the error to propagate to the user, and we want to avoid copying unless
256     # we absolutely have to.
257     except ValueError:
258         result = func(partition.copy(), **kwargs)
259     return (
260         result,
261         len(result) if hasattr(result, "__len__") else 0,
262         len(result.columns) if hasattr(result, "columns") else 0,
263         get_node_ip_address(),
264     )
265
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/modin/engines/ray/pandas_on_ray/frame/partition.py b/modin/engines/ray/pandas_on_ray/frame/partition.py
--- a/modin/engines/ray/pandas_on_ray/frame/partition.py
+++ b/modin/engines/ray/pandas_on_ray/frame/partition.py
@@ -20,11 +20,18 @@
 import ray
 from ray.worker import RayTaskError
 from ray.services import get_node_ip_address
+from packaging import version
+
+ObjectIDType = ray.ObjectRef
+if version.parse(ray.__version__) >= version.parse("1.2.0"):
+    from ray.util.client.common import ClientObjectRef
+
+    ObjectIDType = (ray.ObjectRef, ClientObjectRef)
 
 
 class PandasOnRayFramePartition(BaseFramePartition):
     def __init__(self, object_id, length=None, width=None, ip=None, call_queue=None):
-        assert type(object_id) is ray.ObjectRef
+        assert isinstance(object_id, ObjectIDType)
 
         self.oid = object_id
         if call_queue is None:
@@ -178,7 +185,7 @@
                 self._length_cache, self._width_cache = get_index_and_columns.remote(
                     self.oid
                 )
-        if isinstance(self._length_cache, ray.ObjectRef):
+        if isinstance(self._length_cache, ObjectIDType):
             try:
                 self._length_cache = ray.get(self._length_cache)
             except RayTaskError as e:
@@ -193,7 +200,7 @@
                 self._length_cache, self._width_cache = get_index_and_columns.remote(
                     self.oid
                 )
-        if isinstance(self._width_cache, ray.ObjectRef):
+        if isinstance(self._width_cache, ObjectIDType):
             try:
                 self._width_cache = ray.get(self._width_cache)
             except RayTaskError as e:
@@ -206,7 +213,7 @@
                 self.drain_call_queue()
             else:
                 self._ip_cache = self.apply(lambda df: df)._ip_cache
-        if isinstance(self._ip_cache, ray.ObjectID):
+        if isinstance(self._ip_cache, ObjectIDType):
             try:
                 self._ip_cache = ray.get(self._ip_cache)
             except RayTaskError as e:
@@ -234,7 +241,7 @@
 @ray.remote(num_returns=4)
 def deploy_ray_func(call_queue, partition):  # pragma: no cover
     def deserialize(obj):
-        if isinstance(obj, ray.ObjectRef):
+        if isinstance(obj, ObjectIDType):
             return ray.get(obj)
         return obj

{"golden_diff": "diff --git a/modin/engines/ray/pandas_on_ray/frame/partition.py b/modin/engines/ray/pandas_on_ray/frame/partition.py\n--- a/modin/engines/ray/pandas_on_ray/frame/partition.py\n+++ b/modin/engines/ray/pandas_on_ray/frame/partition.py\n@@ -20,11 +20,18 @@\n import ray\n from ray.worker import RayTaskError\n from ray.services import get_node_ip_address\n+from packaging import version\n+\n+ObjectIDType = ray.ObjectRef\n+if version.parse(ray.__version__) >= version.parse(\"1.2.0\"):\n+    from ray.util.client.common import ClientObjectRef\n+\n+    ObjectIDType = (ray.ObjectRef, ClientObjectRef)\n \n \n class PandasOnRayFramePartition(BaseFramePartition):\n     def __init__(self, object_id, length=None, width=None, ip=None, call_queue=None):\n-        assert type(object_id) is ray.ObjectRef\n+        assert isinstance(object_id, ObjectIDType)\n \n         self.oid = object_id\n         if call_queue is None:\n@@ -178,7 +185,7 @@\n                 self._length_cache, self._width_cache = get_index_and_columns.remote(\n                     self.oid\n                 )\n-        if isinstance(self._length_cache, ray.ObjectRef):\n+        if isinstance(self._length_cache, ObjectIDType):\n             try:\n                 self._length_cache = ray.get(self._length_cache)\n             except RayTaskError as e:\n@@ -193,7 +200,7 @@\n                 self._length_cache, self._width_cache = get_index_and_columns.remote(\n                     self.oid\n                 )\n-        if isinstance(self._width_cache, ray.ObjectRef):\n+        if isinstance(self._width_cache, ObjectIDType):\n             try:\n                 self._width_cache = ray.get(self._width_cache)\n             except RayTaskError as e:\n@@ -206,7 +213,7 @@\n                 self.drain_call_queue()\n             else:\n                 self._ip_cache = self.apply(lambda df: df)._ip_cache\n-        if isinstance(self._ip_cache, ray.ObjectID):\n+        if isinstance(self._ip_cache, ObjectIDType):\n             try:\n                 self._ip_cache = ray.get(self._ip_cache)\n             except RayTaskError as e:\n@@ -234,7 +241,7 @@\n @ray.remote(num_returns=4)\n def deploy_ray_func(call_queue, partition):  # pragma: no cover\n     def deserialize(obj):\n-        if isinstance(obj, ray.ObjectRef):\n+        if isinstance(obj, ObjectIDType):\n             return ray.get(obj)\n         return obj\n", "issue": "Fix defensive assertions for Ray Client users\nRelated to https://github.com/ray-project/ray/issues/14042\r\n\r\nFor now, we will check for both types.\n", "before_files": [{"content": "# Licensed to Modin Development Team under one or more contributor license agreements.\n# See the NOTICE file distributed with this work for additional information regarding\n# copyright ownership.  The Modin Development Team licenses this file to you under the\n# Apache License, Version 2.0 (the \"License\"); you may not use this file except in\n# compliance with the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software distributed under\n# the License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF\n# ANY KIND, either express or implied. See the License for the specific language\n# governing permissions and limitations under the License.\n\nimport pandas\n\nfrom modin.data_management.utils import length_fn_pandas, width_fn_pandas\nfrom modin.engines.base.frame.partition import BaseFramePartition\nfrom modin.engines.ray.utils import handle_ray_task_error\n\nimport ray\nfrom ray.worker import RayTaskError\nfrom ray.services import get_node_ip_address\n\n\nclass PandasOnRayFramePartition(BaseFramePartition):\n    def __init__(self, object_id, length=None, width=None, ip=None, call_queue=None):\n        assert type(object_id) is ray.ObjectRef\n\n        self.oid = object_id\n        if call_queue is None:\n            call_queue = []\n        self.call_queue = call_queue\n        self._length_cache = length\n        self._width_cache = width\n        self._ip_cache = ip\n\n    def get(self):\n        \"\"\"Gets the object out of the plasma store.\n\n        Returns:\n            The object from the plasma store.\n        \"\"\"\n        if len(self.call_queue):\n            self.drain_call_queue()\n        try:\n            return ray.get(self.oid)\n        except RayTaskError as e:\n            handle_ray_task_error(e)\n\n    def apply(self, func, **kwargs):\n        \"\"\"Apply a function to the object stored in this partition.\n\n        Note: It does not matter if func is callable or an ObjectRef. Ray will\n            handle it correctly either way. The keyword arguments are sent as a\n            dictionary.\n\n        Args:\n            func: The function to apply.\n\n        Returns:\n            A RayRemotePartition object.\n        \"\"\"\n        oid = self.oid\n        call_queue = self.call_queue + [(func, kwargs)]\n        result, length, width, ip = deploy_ray_func.remote(call_queue, oid)\n        return PandasOnRayFramePartition(result, length, width, ip)\n\n    def add_to_apply_calls(self, func, **kwargs):\n        return PandasOnRayFramePartition(\n            self.oid, call_queue=self.call_queue + [(func, kwargs)]\n        )\n\n    def drain_call_queue(self):\n        if len(self.call_queue) == 0:\n            return\n        oid = self.oid\n        call_queue = self.call_queue\n        (\n            self.oid,\n            self._length_cache,\n            self._width_cache,\n            self._ip_cache,\n        ) = deploy_ray_func.remote(call_queue, oid)\n        self.call_queue = []\n\n    def wait(self):\n        self.drain_call_queue()\n        try:\n            ray.wait([self.oid])\n        except RayTaskError as e:\n            handle_ray_task_error(e)\n\n    def __copy__(self):\n        return PandasOnRayFramePartition(\n            self.oid,\n            length=self._length_cache,\n            width=self._width_cache,\n            ip=self._ip_cache,\n            call_queue=self.call_queue,\n        )\n\n    def to_pandas(self):\n        \"\"\"Convert the object stored in this partition to a Pandas DataFrame.\n\n        Returns:\n            A Pandas DataFrame.\n        \"\"\"\n        dataframe = self.get()\n        assert type(dataframe) is pandas.DataFrame or type(dataframe) is pandas.Series\n        return dataframe\n\n    def to_numpy(self, **kwargs):\n        \"\"\"\n        Convert the object stored in this partition to a NumPy array.\n\n        Returns\n        -------\n            A NumPy array.\n        \"\"\"\n        return self.apply(lambda df, **kwargs: df.to_numpy(**kwargs)).get()\n\n    def mask(self, row_indices, col_indices):\n        if (\n            (isinstance(row_indices, slice) and row_indices == slice(None))\n            or (\n                not isinstance(row_indices, slice)\n                and self._length_cache is not None\n                and len(row_indices) == self._length_cache\n            )\n        ) and (\n            (isinstance(col_indices, slice) and col_indices == slice(None))\n            or (\n                not isinstance(col_indices, slice)\n                and self._width_cache is not None\n                and len(col_indices) == self._width_cache\n            )\n        ):\n            return self.__copy__()\n\n        new_obj = self.add_to_apply_calls(\n            lambda df: pandas.DataFrame(df.iloc[row_indices, col_indices])\n        )\n        if not isinstance(row_indices, slice):\n            new_obj._length_cache = len(row_indices)\n        if not isinstance(col_indices, slice):\n            new_obj._width_cache = len(col_indices)\n        return new_obj\n\n    @classmethod\n    def put(cls, obj):\n        \"\"\"Put an object in the Plasma store and wrap it in this object.\n\n        Args:\n            obj: The object to be put.\n\n        Returns:\n            A `RayRemotePartition` object.\n        \"\"\"\n        return PandasOnRayFramePartition(ray.put(obj), len(obj.index), len(obj.columns))\n\n    @classmethod\n    def preprocess_func(cls, func):\n        \"\"\"Put a callable function into the plasma store for use in `apply`.\n\n        Args:\n            func: The function to preprocess.\n\n        Returns:\n            A ray.ObjectRef.\n        \"\"\"\n        return ray.put(func)\n\n    def length(self):\n        if self._length_cache is None:\n            if len(self.call_queue):\n                self.drain_call_queue()\n            else:\n                self._length_cache, self._width_cache = get_index_and_columns.remote(\n                    self.oid\n                )\n        if isinstance(self._length_cache, ray.ObjectRef):\n            try:\n                self._length_cache = ray.get(self._length_cache)\n            except RayTaskError as e:\n                handle_ray_task_error(e)\n        return self._length_cache\n\n    def width(self):\n        if self._width_cache is None:\n            if len(self.call_queue):\n                self.drain_call_queue()\n            else:\n                self._length_cache, self._width_cache = get_index_and_columns.remote(\n                    self.oid\n                )\n        if isinstance(self._width_cache, ray.ObjectRef):\n            try:\n                self._width_cache = ray.get(self._width_cache)\n            except RayTaskError as e:\n                handle_ray_task_error(e)\n        return self._width_cache\n\n    def ip(self):\n        if self._ip_cache is None:\n            if len(self.call_queue):\n                self.drain_call_queue()\n            else:\n                self._ip_cache = self.apply(lambda df: df)._ip_cache\n        if isinstance(self._ip_cache, ray.ObjectID):\n            try:\n                self._ip_cache = ray.get(self._ip_cache)\n            except RayTaskError as e:\n                handle_ray_task_error(e)\n        return self._ip_cache\n\n    @classmethod\n    def length_extraction_fn(cls):\n        return length_fn_pandas\n\n    @classmethod\n    def width_extraction_fn(cls):\n        return width_fn_pandas\n\n    @classmethod\n    def empty(cls):\n        return cls.put(pandas.DataFrame())\n\n\n@ray.remote(num_returns=2)\ndef get_index_and_columns(df):\n    return len(df.index), len(df.columns)\n\n\n@ray.remote(num_returns=4)\ndef deploy_ray_func(call_queue, partition):  # pragma: no cover\n    def deserialize(obj):\n        if isinstance(obj, ray.ObjectRef):\n            return ray.get(obj)\n        return obj\n\n    if len(call_queue) > 1:\n        for func, kwargs in call_queue[:-1]:\n            func = deserialize(func)\n            kwargs = deserialize(kwargs)\n            try:\n                partition = func(partition, **kwargs)\n            except ValueError:\n                partition = func(partition.copy(), **kwargs)\n    func, kwargs = call_queue[-1]\n    func = deserialize(func)\n    kwargs = deserialize(kwargs)\n    try:\n        result = func(partition, **kwargs)\n    # Sometimes Arrow forces us to make a copy of an object before we operate on it. We\n    # don't want the error to propagate to the user, and we want to avoid copying unless\n    # we absolutely have to.\n    except ValueError:\n        result = func(partition.copy(), **kwargs)\n    return (\n        result,\n        len(result) if hasattr(result, \"__len__\") else 0,\n        len(result.columns) if hasattr(result, \"columns\") else 0,\n        get_node_ip_address(),\n    )\n", "path": "modin/engines/ray/pandas_on_ray/frame/partition.py"}], "after_files": [{"content": "# Licensed to Modin Development Team under one or more contributor license agreements.\n# See the NOTICE file distributed with this work for additional information regarding\n# copyright ownership.  The Modin Development Team licenses this file to you under the\n# Apache License, Version 2.0 (the \"License\"); you may not use this file except in\n# compliance with the License.  You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software distributed under\n# the License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF\n# ANY KIND, either express or implied. See the License for the specific language\n# governing permissions and limitations under the License.\n\nimport pandas\n\nfrom modin.data_management.utils import length_fn_pandas, width_fn_pandas\nfrom modin.engines.base.frame.partition import BaseFramePartition\nfrom modin.engines.ray.utils import handle_ray_task_error\n\nimport ray\nfrom ray.worker import RayTaskError\nfrom ray.services import get_node_ip_address\nfrom packaging import version\n\nObjectIDType = ray.ObjectRef\nif version.parse(ray.__version__) >= version.parse(\"1.2.0\"):\n    from ray.util.client.common import ClientObjectRef\n\n    ObjectIDType = (ray.ObjectRef, ClientObjectRef)\n\n\nclass PandasOnRayFramePartition(BaseFramePartition):\n    def __init__(self, object_id, length=None, width=None, ip=None, call_queue=None):\n        assert isinstance(object_id, ObjectIDType)\n\n        self.oid = object_id\n        if call_queue is None:\n            call_queue = []\n        self.call_queue = call_queue\n        self._length_cache = length\n        self._width_cache = width\n        self._ip_cache = ip\n\n    def get(self):\n        \"\"\"Gets the object out of the plasma store.\n\n        Returns:\n            The object from the plasma store.\n        \"\"\"\n        if len(self.call_queue):\n            self.drain_call_queue()\n        try:\n            return ray.get(self.oid)\n        except RayTaskError as e:\n            handle_ray_task_error(e)\n\n    def apply(self, func, **kwargs):\n        \"\"\"Apply a function to the object stored in this partition.\n\n        Note: It does not matter if func is callable or an ObjectRef. Ray will\n            handle it correctly either way. The keyword arguments are sent as a\n            dictionary.\n\n        Args:\n            func: The function to apply.\n\n        Returns:\n            A RayRemotePartition object.\n        \"\"\"\n        oid = self.oid\n        call_queue = self.call_queue + [(func, kwargs)]\n        result, length, width, ip = deploy_ray_func.remote(call_queue, oid)\n        return PandasOnRayFramePartition(result, length, width, ip)\n\n    def add_to_apply_calls(self, func, **kwargs):\n        return PandasOnRayFramePartition(\n            self.oid, call_queue=self.call_queue + [(func, kwargs)]\n        )\n\n    def drain_call_queue(self):\n        if len(self.call_queue) == 0:\n            return\n        oid = self.oid\n        call_queue = self.call_queue\n        (\n            self.oid,\n            self._length_cache,\n            self._width_cache,\n            self._ip_cache,\n        ) = deploy_ray_func.remote(call_queue, oid)\n        self.call_queue = []\n\n    def wait(self):\n        self.drain_call_queue()\n        try:\n            ray.wait([self.oid])\n        except RayTaskError as e:\n            handle_ray_task_error(e)\n\n    def __copy__(self):\n        return PandasOnRayFramePartition(\n            self.oid,\n            length=self._length_cache,\n            width=self._width_cache,\n            ip=self._ip_cache,\n            call_queue=self.call_queue,\n        )\n\n    def to_pandas(self):\n        \"\"\"Convert the object stored in this partition to a Pandas DataFrame.\n\n        Returns:\n            A Pandas DataFrame.\n        \"\"\"\n        dataframe = self.get()\n        assert type(dataframe) is pandas.DataFrame or type(dataframe) is pandas.Series\n        return dataframe\n\n    def to_numpy(self, **kwargs):\n        \"\"\"\n        Convert the object stored in this partition to a NumPy array.\n\n        Returns\n        -------\n            A NumPy array.\n        \"\"\"\n        return self.apply(lambda df, **kwargs: df.to_numpy(**kwargs)).get()\n\n    def mask(self, row_indices, col_indices):\n        if (\n            (isinstance(row_indices, slice) and row_indices == slice(None))\n            or (\n                not isinstance(row_indices, slice)\n                and self._length_cache is not None\n                and len(row_indices) == self._length_cache\n            )\n        ) and (\n            (isinstance(col_indices, slice) and col_indices == slice(None))\n            or (\n                not isinstance(col_indices, slice)\n                and self._width_cache is not None\n                and len(col_indices) == self._width_cache\n            )\n        ):\n            return self.__copy__()\n\n        new_obj = self.add_to_apply_calls(\n            lambda df: pandas.DataFrame(df.iloc[row_indices, col_indices])\n        )\n        if not isinstance(row_indices, slice):\n            new_obj._length_cache = len(row_indices)\n        if not isinstance(col_indices, slice):\n            new_obj._width_cache = len(col_indices)\n        return new_obj\n\n    @classmethod\n    def put(cls, obj):\n        \"\"\"Put an object in the Plasma store and wrap it in this object.\n\n        Args:\n            obj: The object to be put.\n\n        Returns:\n            A `RayRemotePartition` object.\n        \"\"\"\n        return PandasOnRayFramePartition(ray.put(obj), len(obj.index), len(obj.columns))\n\n    @classmethod\n    def preprocess_func(cls, func):\n        \"\"\"Put a callable function into the plasma store for use in `apply`.\n\n        Args:\n            func: The function to preprocess.\n\n        Returns:\n            A ray.ObjectRef.\n        \"\"\"\n        return ray.put(func)\n\n    def length(self):\n        if self._length_cache is None:\n            if len(self.call_queue):\n                self.drain_call_queue()\n            else:\n                self._length_cache, self._width_cache = get_index_and_columns.remote(\n                    self.oid\n                )\n        if isinstance(self._length_cache, ObjectIDType):\n            try:\n                self._length_cache = ray.get(self._length_cache)\n            except RayTaskError as e:\n                handle_ray_task_error(e)\n        return self._length_cache\n\n    def width(self):\n        if self._width_cache is None:\n            if len(self.call_queue):\n                self.drain_call_queue()\n            else:\n                self._length_cache, self._width_cache = get_index_and_columns.remote(\n                    self.oid\n                )\n        if isinstance(self._width_cache, ObjectIDType):\n            try:\n                self._width_cache = ray.get(self._width_cache)\n            except RayTaskError as e:\n                handle_ray_task_error(e)\n        return self._width_cache\n\n    def ip(self):\n        if self._ip_cache is None:\n            if len(self.call_queue):\n                self.drain_call_queue()\n            else:\n                self._ip_cache = self.apply(lambda df: df)._ip_cache\n        if isinstance(self._ip_cache, ObjectIDType):\n            try:\n                self._ip_cache = ray.get(self._ip_cache)\n            except RayTaskError as e:\n                handle_ray_task_error(e)\n        return self._ip_cache\n\n    @classmethod\n    def length_extraction_fn(cls):\n        return length_fn_pandas\n\n    @classmethod\n    def width_extraction_fn(cls):\n        return width_fn_pandas\n\n    @classmethod\n    def empty(cls):\n        return cls.put(pandas.DataFrame())\n\n\n@ray.remote(num_returns=2)\ndef get_index_and_columns(df):\n    return len(df.index), len(df.columns)\n\n\n@ray.remote(num_returns=4)\ndef deploy_ray_func(call_queue, partition):  # pragma: no cover\n    def deserialize(obj):\n        if isinstance(obj, ObjectIDType):\n            return ray.get(obj)\n        return obj\n\n    if len(call_queue) > 1:\n        for func, kwargs in call_queue[:-1]:\n            func = deserialize(func)\n            kwargs = deserialize(kwargs)\n            try:\n                partition = func(partition, **kwargs)\n            except ValueError:\n                partition = func(partition.copy(), **kwargs)\n    func, kwargs = call_queue[-1]\n    func = deserialize(func)\n    kwargs = deserialize(kwargs)\n    try:\n        result = func(partition, **kwargs)\n    # Sometimes Arrow forces us to make a copy of an object before we operate on it. We\n    # don't want the error to propagate to the user, and we want to avoid copying unless\n    # we absolutely have to.\n    except ValueError:\n        result = func(partition.copy(), **kwargs)\n    return (\n        result,\n        len(result) if hasattr(result, \"__len__\") else 0,\n        len(result.columns) if hasattr(result, \"columns\") else 0,\n        get_node_ip_address(),\n    )\n", "path": "modin/engines/ray/pandas_on_ray/frame/partition.py"}]}