[data] Always convert arrow batches to pandas batches when user specifies batch_format="native"
diff --git a/python/ray/data/block.py b/python/ray/data/block.py
index 7cf54d4219794..0a83e17dd38f4 100644
--- a/python/ray/data/block.py
+++ b/python/ray/data/block.py
@@ -279,7 +279,7 @@ def for_block(block: Block) -> "BlockAccessor[T]":
 
             return SimpleBlockAccessor(block)
         else:
-            raise TypeError("Not a block type: {}".format(block))
+            raise TypeError("Not a block type: {} ({})".format(block, type(block)))
 
     def sample(self, n_samples: int, key: Any) -> "Block[T]":
         """Return a random sample of items from this block."""
diff --git a/python/ray/data/dataset.py b/python/ray/data/dataset.py
index 07bfabb34cee6..84bb57b43ecd8 100644
--- a/python/ray/data/dataset.py
+++ b/python/ray/data/dataset.py
@@ -221,8 +221,9 @@ def map_batches(
                 blocks as batches. Defaults to a system-chosen batch size.
             compute: The compute strategy, either "tasks" (default) to use Ray
                 tasks, or "actors" to use an autoscaling Ray actor pool.
-            batch_format: Specify "native" to use the native block format,
-                "pandas" to select ``pandas.DataFrame`` as the batch format,
+            batch_format: Specify "native" to use the native block format
+                (promotes Arrow to pandas), "pandas" to select
+                ``pandas.DataFrame`` as the batch format,
                 or "pyarrow" to select ``pyarrow.Table``.
             ray_remote_args: Additional resource requirements to request from
                 ray (e.g., num_gpus=1 to request GPUs for the map tasks).
@@ -252,7 +253,9 @@ def transform(block: Block) -> Iterable[Block]:
                 # bug where we include the entire base view on serialization.
                 view = block.slice(start, end, copy=batch_size is not None)
                 if batch_format == "native":
-                    pass
+                    # Always promote Arrow blocks to pandas for consistency.
+                    if isinstance(view, pa.Table) or isinstance(view, bytes):
+                        view = BlockAccessor.for_block(view).to_pandas()
                 elif batch_format == "pandas":
                     view = BlockAccessor.for_block(view).to_pandas()
                 elif batch_format == "pyarrow":
@@ -1831,7 +1834,8 @@ def iter_batches(
                 current block during the scan.
             batch_size: Record batch size, or None to let the system pick.
             batch_format: The format in which to return each batch.
-                Specify "native" to use the current block format, "pandas" to
+                Specify "native" to use the current block format (promoting
+                Arrow to pandas automatically), "pandas" to
                 select ``pandas.DataFrame`` or "pyarrow" to select
                 ``pyarrow.Table``. Default is "native".
             drop_last: Whether to drop the last batch if it's incomplete.
@@ -1840,10 +1844,17 @@ def iter_batches(
             A list of iterators over record batches.
         """
 
+        import pyarrow as pa
+
         time_start = time.perf_counter()
 
         def format_batch(batch: Block, format: str) -> BatchType:
             if batch_format == "native":
+                # Always promote Arrow blocks to pandas for consistency, since
+                # we lazily convert pandas->Arrow internally for efficiency.
+                if isinstance(batch, pa.Table) or isinstance(batch, bytes):
+                    batch = BlockAccessor.for_block(batch)
+                    batch = batch.to_pandas()
                 return batch
             elif batch_format == "pandas":
                 batch = BlockAccessor.for_block(batch)
diff --git a/python/ray/data/impl/pandas_block.py b/python/ray/data/impl/pandas_block.py
index 915dd7fd04819..76279df05053c 100644
--- a/python/ray/data/impl/pandas_block.py
+++ b/python/ray/data/impl/pandas_block.py
@@ -32,7 +32,6 @@ def as_pydict(self) -> dict:
         return {k: v[0] for k, v in self._row.to_dict("list").items()}
 
     def __getitem__(self, key: str) -> Any:
-        assert isinstance(key, str)
         col = self._row[key]
         if len(col) == 0:
             return None
diff --git a/python/ray/data/tests/test_dataset.py b/python/ray/data/tests/test_dataset.py
index 9e468b98fbc0f..d3532c680a262 100644
--- a/python/ray/data/tests/test_dataset.py
+++ b/python/ray/data/tests/test_dataset.py
@@ -1230,10 +1230,8 @@ def test_from_numpy(ray_start_regular_shared):
     arr1 = np.expand_dims(np.arange(0, 4), axis=1)
     arr2 = np.expand_dims(np.arange(4, 8), axis=1)
     ds = ray.data.from_numpy([ray.put(arr1), ray.put(arr2)])
-    values = np.array(ds.take(8))
-    np.testing.assert_array_equal(
-        values, np.expand_dims(np.concatenate((arr1, arr2)), axis=1)
-    )
+    values = np.stack([x["value"] for x in ds.take(8)])
+    np.testing.assert_array_equal(values, np.concatenate((arr1, arr2)))
 
 
 def test_from_arrow(ray_start_regular_shared):

<!-- 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?

With the addition of https://github.com/ray-project/ray/pull/20988, the native format becomes ambiguous. This PR proposes to auto-promote arrow to pandas blocks when the user specifies "native" format, to avoid uncertainty.