code: def test_tensors_in_tables_parquet_pickle_manual_serde(
      ray_start_regular_shared, tmp_path
  ):
      import pickle
  
      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": [pickle.dumps(a) for a in arr],
          }
      )
      ds = ray.data.from_pandas([df])
      ds.write_parquet(str(tmp_path))
      ds = ray.data.read_parquet(str(tmp_path))
  
      # Manually deserialize the tensor pickle bytes and cast to our tensor
      # extension type.
      def deser_mapper(batch: pd.DataFrame):
          batch["two"] = [pickle.loads(a) for a in batch["two"]]
          batch["two"] = batch["two"].astype(
              TensorDtype(shape, np.int64)
          )
          return batch
  
      casted_ds = ds.map_batches(deser_mapper, batch_format="pandas")
  
      values = [[s["one"], s["two"]] for s in casted_ds.take()]
      expected = list(zip(list(range(outer_dim)), arr))
      for v, e in zip(sorted(values), expected):
          np.testing.assert_equal(v, e)
  
      # Manually deserialize the pickle tensor bytes and directly cast it to a
      # TensorArray.
      def deser_mapper_direct(batch: pd.DataFrame):
          batch["two"] = TensorArray(
              [pickle.loads(a) for a in batch["two"]]
          )
          return batch
  
      casted_ds = ds.map_batches(
          deser_mapper_direct, batch_format="pandas"
      )
  
      values = [[s["one"], s["two"]] for s in casted_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 16 for calling a global function with example usage: test_tensors_in_tables_parquet_pickle_manual_serde(ray_start_regular_shared, tmp_path) and returns: batch, batch