code: def test_tensors_in_tables_parquet_bytes_manual_serde(
      ray_start_regular_shared, tmp_path
  ):
      outer_dim = 3
      inner_shape = (2, 2, 2)
      shape = (outer_dim,) + inner_shape
      num_items = np.prod(np.array(shape))
      arr = np.arange(num_items).reshape(shape)
      df = pd.DataFrame(
          {
              "one": list(range(outer_dim)),
              "two": [a.tobytes() for a in arr],
          }
      )
      ds = ray.data.from_pandas([df])
      ds.write_parquet(str(tmp_path))
      ds = ray.data.read_parquet(str(tmp_path))
  
      tensor_col_name = "two"
  
      # Manually deserialize the tensor bytes and cast to a TensorArray.
      def np_deser_mapper(batch: pa.Table):
          # NOTE(Clark): We use NumPy to consolidate these potentially
          # non-contiguous buffers, and to do buffer bookkeeping in general.
          np_col = np.array(
              [
                  np.ndarray(
                      inner_shape,
                      buffer=buf.as_buffer(),
                      dtype=arr.dtype,
                  )
                  for buf in batch.column(tensor_col_name)
              ]
          )
  
          return batch.set_column(
              batch._ensure_integer_index(tensor_col_name),
              tensor_col_name,
              ArrowTensorArray.from_numpy(np_col),
          )
  
      ds = ds.map_batches(np_deser_mapper, batch_format="pyarrow")
  
      values = [[s["one"], s["two"]] for s in ds.take()]
      expected = list(zip(list(range(outer_dim)), arr))
      for v, e in zip(sorted(values), expected):
          np.testing.assert_equal(v, e)
  
imports: import time
  import numpy
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.air.util.tensor_extensions.utils import _create_possibly_ragged_ndarray
  from ray.data.block import BlockAccessor
  from ray.data.extensions.tensor_extension import ArrowTensorArray, ArrowTensorType, ArrowVariableShapedTensorArray, ArrowVariableShapedTensorType, TensorArray, TensorDtype
  from ray.data.tests.conftest import *
  from ray.data.tests.util import extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 17 for calling a global function with example usage: test_tensors_in_tables_parquet_bytes_manual_serde(ray_start_regular_shared, tmp_path) and returns: batch