PERF-#0000: call 'to_numpy' in main process

Signed-off-by: Anatoly Myachev <[email protected]>

modin/core/execution/ray/generic/partitioning/partition_manager.py

@@ -41,14 +41,9 @@ def to_numpy(cls, partitions, **kwargs):
         NumPy array
         """
         parts = RayWrapper.materialize(
-            [
-                obj.apply(lambda df, **kwargs: df.to_numpy(**kwargs)).list_of_blocks[0]
-                for row in partitions
-                for obj in row
-            ]
+            [obj.list_of_blocks[0] for row in partitions for obj in row]
         )
-        n = partitions.shape[1]
-        parts = [parts[i * n : (i + 1) * n] for i in list(range(partitions.shape[0]))]
-
-        arr = np.block(parts)
-        return arr
+        parts = [part.to_numpy() for part in parts]
+        rows, cols = partitions.shape
+        parts = [parts[i * cols : (i + 1) * cols] for i in range(rows)]
+        return np.block(parts)

modin/core/execution/unidist/generic/partitioning/partition_manager.py

@@ -41,12 +41,9 @@ def to_numpy(cls, partitions, **kwargs):
         NumPy array
         """
         parts = UnidistWrapper.materialize(
-            [
-                obj.apply(lambda df, **kwargs: df.to_numpy(**kwargs)).list_of_blocks[0]
-                for row in partitions
-                for obj in row
-            ]
+            [obj.list_of_blocks[0] for row in partitions for obj in row]
         )
+        parts = [part.to_numpy() for part in parts]
         rows, cols = partitions.shape
         parts = [parts[i * cols : (i + 1) * cols] for i in range(rows)]
         return np.block(parts)