code: def test_tensors_basic(ray_start_regular_shared):
      # Create directly.
      tensor_shape = (3, 5)
      ds = ray.data.range_tensor(6, shape=tensor_shape, parallelism=6)
      assert str(ds) == (
          "Dataset(\n"
          "   num_blocks=6,\n"
          "   num_rows=6,\n"
          "   schema={data: numpy.ndarray(shape=(3, 5), dtype=int64)}\n"
          ")"
      )
      assert ds.size_bytes() == 5 * 3 * 6 * 8
  
      # Test row iterator yields tensors.
      for tensor in ds.iter_rows():
          tensor = tensor["data"]
          assert isinstance(tensor, np.ndarray)
          assert tensor.shape == tensor_shape
  
      # Test batch iterator yields tensors.
      for tensor in ds.iter_batches(batch_size=2):
          tensor = tensor["data"]
          assert isinstance(tensor, np.ndarray)
          assert tensor.shape == (2,) + tensor_shape
  
      # Native format.
      def np_mapper(arr):
          if "data" in arr:
              arr = arr["data"]
          else:
              arr = arr["id"]
          assert isinstance(arr, np.ndarray)
          return {"data": arr + 1}
  
      res = ray.data.range_tensor(2, shape=(2, 2)).map(np_mapper).take()
      np.testing.assert_equal(
          extract_values("data", res),
          [np.ones((2, 2)), 2 * np.ones((2, 2))],
      )
  
      # Explicit NumPy format.
      res = (
          ray.data.range_tensor(2, shape=(2, 2))
          .map_batches(np_mapper, batch_format="numpy")
          .take()
      )
      np.testing.assert_equal(
          extract_values("data", res),
          [np.ones((2, 2)), 2 * np.ones((2, 2))],
      )
  
      # Pandas conversion.
      def pd_mapper(df):
          assert isinstance(df, pd.DataFrame)
          return df + 2
  
      res = (
          ray.data.range_tensor(2)
          .map_batches(pd_mapper, batch_format="pandas")
          .take()
      )
      np.testing.assert_equal(
          extract_values("data", res), [np.array([2]), np.array([3])]
      )
  
      # Arrow columns in NumPy format.
      def multi_mapper(col_arrs):
          assert isinstance(col_arrs, dict)
          assert list(col_arrs.keys()) == ["a", "b", "c"]
          assert all(
              isinstance(col_arr, np.ndarray)
              for col_arr in col_arrs.values()
          )
          return {
              "a": col_arrs["a"] + 1,
              "b": col_arrs["b"] + 1,
              "c": col_arrs["c"] + 1,
          }
  
      # Multiple columns.
      t = pa.table(
          {
              "a": [1, 2, 3],
              "b": [4.0, 5.0, 6.0],
              "c": ArrowTensorArray.from_numpy(
                  np.array([[1, 2], [3, 4], [5, 6]])
              ),
          }
      )
      res = (
          ray.data.from_arrow(t)
          .map_batches(multi_mapper, batch_size=2, batch_format="numpy")
          .take()
      )
      np.testing.assert_equal(
          res,
          [
              {"a": 2, "b": 5.0, "c": np.array([2, 3])},
              {"a": 3, "b": 6.0, "c": np.array([4, 5])},
              {"a": 4, "b": 7.0, "c": np.array([6, 7])},
          ],
      )
  
      def single_mapper(col_arrs):
          assert isinstance(col_arrs, dict)
          assert list(col_arrs.keys()) == ["c"]
          assert all(
              isinstance(col_arr, np.ndarray)
              for col_arr in col_arrs.values()
          )
          return {"c": col_arrs["c"] + 1}
  
      # Single column (should still yield ndarray dict batches).
      t = t.select(["c"])
      res = (
          ray.data.from_arrow(t)
          .map_batches(
              single_mapper, batch_size=2, batch_format="numpy"
          )
          .take()
      )
      np.testing.assert_equal(
          res,
          [
              {"c": np.array([2, 3])},
              {"c": np.array([4, 5])},
              {"c": np.array([6, 7])},
          ],
      )
  
      # Pandas columns in NumPy format.
      def multi_mapper(col_arrs):
          assert isinstance(col_arrs, dict)
          assert list(col_arrs.keys()) == ["a", "b", "c"]
          assert all(
              isinstance(col_arr, np.ndarray)
              for col_arr in col_arrs.values()
          )
          return pd.DataFrame(
              {
                  "a": col_arrs["a"] + 1,
                  "b": col_arrs["b"] + 1,
                  "c": TensorArray(col_arrs["c"] + 1),
              }
          )
  
      # Multiple columns.
      df = pd.DataFrame(
          {
              "a": [1, 2, 3],
              "b": [4.0, 5.0, 6.0],
              "c": TensorArray(np.array([[1, 2], [3, 4], [5, 6]])),
          }
      )
      res = (
          ray.data.from_pandas(df)
          .map_batches(multi_mapper, batch_size=2, batch_format="numpy")
          .take()
      )
      np.testing.assert_equal(
          res,
          [
              {"a": 2, "b": 5.0, "c": np.array([2, 3])},
              {"a": 3, "b": 6.0, "c": np.array([4, 5])},
              {"a": 4, "b": 7.0, "c": np.array([6, 7])},
          ],
      )
  
      # Single column (should still yield ndarray dict batches).
      def single_mapper(col_arrs):
          assert isinstance(col_arrs, dict)
          assert list(col_arrs.keys()) == ["c"]
          assert all(
              isinstance(col_arr, np.ndarray)
              for col_arr in col_arrs.values()
          )
          return pd.DataFrame({"c": TensorArray(col_arrs["c"] + 1)})
  
      df = df[["c"]]
      res = (
          ray.data.from_pandas(df)
          .map_batches(
              single_mapper, batch_size=2, batch_format="numpy"
          )
          .take()
      )
      np.testing.assert_equal(
          res,
          [
              {"c": np.array([2, 3])},
              {"c": np.array([4, 5])},
              {"c": np.array([6, 7])},
          ],
      )
  
      # Simple dataset in NumPy format.
      def mapper(arr):
          arr = np_mapper(arr)
          return arr
  
      res = (
          ray.data.range(10, parallelism=2)
          .map_batches(mapper, batch_format="numpy")
          .take()
      )
      assert extract_values("data", res) == list(range(1, 11))
  
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 2 for calling a global function with example usage: test_tensors_basic(ray_start_regular_shared) and returns: dict, unknown, dict, dict, pd, pd, arr