test

modin/core/execution/ray/implementations/pandas_on_ray/partitioning/lazy_virtual_partition.py

@@ -212,21 +212,23 @@ def apply(
                 other_part = other_axis_partition._data
             args = [other_part] + list(args)
 
-        de = self._apply(func, args, kwargs)
+        obj, meta, meta_off = self._apply(func, args, kwargs).exec()
         if num_splits > 1:
-            de = _DeferredSplit(de, self.axis, num_splits, lengths)
+            obj = _DeferredSplit(obj, self.axis, num_splits, lengths)
             if lengths is not None and len(lengths) != num_splits:
                 lengths = None
             result = [
                 PandasOnRayDataframePartition(
                     _DeferredGetChunk(
-                        de, i, lengths[i] if lengths is not None else None
+                        obj, i, lengths[i] if lengths is not None else None
                     )
                 )
                 for i in range(num_splits)
             ]
         else:
-            result = [PandasOnRayDataframePartition(de)]
+            result = [
+                PandasOnRayDataframePartition(obj, meta=meta, meta_offset=meta_off)
+            ]
         if self.full_axis or other_axis_partition is not None:
             return result
         else:
@@ -338,7 +340,7 @@ def list_of_block_partitions(self) -> list:
         if self._list_of_block_partitions is not None:
             return self._list_of_block_partitions
 
-        data = self._data_ref
+        data = self._data
         num_splits = self._num_splits
         if num_splits > 1:
             lengths = self._chunk_lengths