code: def test_tensors_in_tables_parquet_bytes_manual_serde_col_schema(
      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)
      tensor_col_name = "two"
      df = pd.DataFrame(
          {
              "one": list(range(outer_dim)),
              tensor_col_name: [a.tobytes() for a in arr],
          }
      )
      ds = ray.data.from_pandas([df])
      ds.write_parquet(str(tmp_path))
  
      def _block_udf(block: pa.Table):
          df = block.to_pandas()
          df[tensor_col_name] += 1
          return pa.Table.from_pandas(df)
  
      ds = ray.data.read_parquet(
          str(tmp_path),
          tensor_column_schema={
              tensor_col_name: (arr.dtype, inner_shape)
          },
          _block_udf=_block_udf,
      )
  
      assert isinstance(
          ds.schema().base_schema.field_by_name(tensor_col_name).type,
          ArrowTensorType,
      )
  
      values = [[s["one"], s["two"]] for s in ds.take()]
      expected = list(zip(list(range(outer_dim)), arr + 1))
      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 19 for calling a global function with example usage: test_tensors_in_tables_parquet_bytes_manual_serde_col_schema(ray_start_regular_shared, tmp_path) and returns: pa