[Data] Avoid unnecessary copying of blocks (ray-project#56569)

<!-- Thank you for your contribution! Please review
https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before
opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR.
If you don't have the access to it, we will shortly find a reviewer and
assign them to your PR. -->

## Why are these changes needed?

These were introduced before current serialization protocol inheriting
from Arrow IPC was implemented therefore making this precautionary
copies obsolete (IPC based serializer will actually do the copying
properly).

## Related issue number

<!-- For example: "Closes ray-project#1234" -->

## Checks

- [ ] I've signed off every commit(by using the -s flag, i.e., `git
commit -s`) in this PR.
- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for
https://docs.ray.io/en/master/.
- [ ] I've added any new APIs to the API Reference. For example, if I
added a
method in Tune, I've added it in `doc/source/tune/api/` under the
           corresponding `.rst` file.
- [ ] I've made sure the tests are passing. Note that there might be a
few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(

---------

Signed-off-by: Alexey Kudinkin <ak@anyscale.com>
Signed-off-by: Zhiqiang Ma <zhiqiang.ma@intel.com>

Author: alexeykudinkin

python/ray/data/_internal/execution/operators/limit_operator.py

@@ -54,7 +54,9 @@ def _add_input_inner(self, refs: RefBundle, input_index: int) -> None:
             else:
                 # Slice the last block.
                 def slice_fn(block, metadata, num_rows) -> Tuple[Block, BlockMetadata]:
-                    block = BlockAccessor.for_block(block).slice(0, num_rows, copy=True)
+                    block = BlockAccessor.for_block(block).slice(
+                        0, num_rows, copy=False
+                    )
                     metadata = copy.deepcopy(metadata)
                     metadata.num_rows = num_rows
                     metadata.size_bytes = BlockAccessor.for_block(block).size_bytes()

python/ray/data/_internal/execution/operators/map_transformer.py

@@ -599,8 +599,5 @@ def __call__(self, blocks: Iterable[Block], ctx: TaskContext) -> Iterable[Block]
             offset = 0
             split_sizes = _splitrange(block.num_rows(), self._additional_split_factor)
             for size in split_sizes:
-                # NOTE: copy=True is needed because this is an output block. If
-                # a block slice is put into the object store, the entire block
-                # will get serialized.
-                yield block.slice(offset, offset + size, copy=True)
+                yield block.slice(offset, offset + size, copy=False)
                 offset += size

python/ray/data/_internal/output_buffer.py

@@ -149,18 +149,7 @@ def next(self) -> Block:
             )
 
         if target_num_rows is not None and target_num_rows < accessor.num_rows():
-            # NOTE: We're maintaining following protocol of slicing underlying block
-            #       into appropriately sized ones:
-            #
-            #         - (Finalized) Target blocks sliced from the original one
-            #           and are *copied* to avoid referencing original blocks
-            #         - Temporary remainder of the block should *NOT* be copied
-            #           such as to avoid repeatedly copying the remainder bytes
-            #           of the block, resulting in O(M * N) total bytes being
-            #           copied, where N is the total number of bytes in the original
-            #           block and M is the number of blocks that will be produced by
-            #           this iterator
-            block = accessor.slice(0, target_num_rows, copy=True)
+            block = accessor.slice(0, target_num_rows, copy=False)
             block_remainder = accessor.slice(
                 target_num_rows, accessor.num_rows(), copy=False
             )