diff --git a/paimon-python/pypaimon/multimodal/query.py b/paimon-python/pypaimon/multimodal/query.py
index 13aa615b8054..bba7073da945 100644
--- a/paimon-python/pypaimon/multimodal/query.py
+++ b/paimon-python/pypaimon/multimodal/query.py
@@ -20,6 +20,7 @@
 
 import pyarrow as pa
 
+from pypaimon.casting.schema_evolution import read_arrow_schema
 from pypaimon.common.where_parser import parse_where_clause
 from pypaimon.multimodal.blob_read import fetch_blob_bodies
 from pypaimon.schema.data_types import is_array_blob_type, is_blob_type, is_map_blob_type
@@ -533,7 +534,7 @@ def _read_global_index_result(self, result):
         added_row_id = projection is None or row_id_name not in projection
         if metadata_only:
             row_ids = pa.array(list(result.results()), type=pa.int64())
-            fields = [pa.field(row_id_name, pa.int64(), nullable=False)] * (len(projection) or 1)
+            fields = [read_arrow_schema([SpecialFields.ROW_ID]).field(row_id_name)] * (len(projection) or 1)
             table = pa.Table.from_arrays(
                 [row_ids] * len(fields), schema=pa.schema(fields))
         else:
diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
index c745db9e8e6c..92ce70995410 100644
--- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py
+++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py
@@ -877,16 +877,11 @@ def _apply_group(group: pa.Table) -> pa.Table:
 def _read_output_schema(table, read_cols: Sequence[str]) -> "pa.Schema":
     """Result schema: each projected column's type plus int64 ``_ROW_ID``, in
     ``read_cols`` order. Shared by the empty-result paths so they can't drift."""
-    from pypaimon.schema.data_types import PyarrowFieldParser
+    from pypaimon.casting.schema_evolution import read_arrow_schema
     from pypaimon.table.special_fields import SpecialFields
 
-    rid = SpecialFields.ROW_ID.name
-    full = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields)
-    # Keep each field's nullability so an empty result matches a non-empty read.
-    return pa.schema([
-        pa.field(rid, pa.int64(), nullable=False) if col == rid else full.field(col)
-        for col in read_cols
-    ])
+    full = read_arrow_schema(table.table_schema.fields + [SpecialFields.ROW_ID])
+    return pa.schema([full.field(col) for col in read_cols])
 
 
 def distributed_read_by_row_id(
diff --git a/paimon-python/pypaimon/tests/binary_row_test.py b/paimon-python/pypaimon/tests/binary_row_test.py
index 5e8420e9eb42..940240ab6cac 100644
--- a/paimon-python/pypaimon/tests/binary_row_test.py
+++ b/paimon-python/pypaimon/tests/binary_row_test.py
@@ -26,6 +26,7 @@
 import pytest
 
 from pypaimon import CatalogFactory, Schema
+from pypaimon.casting.schema_evolution import read_arrow_schema
 from pypaimon.manifest.schema.manifest_entry import ManifestEntry
 from pypaimon.manifest.schema.simple_stats import SimpleStats
 from pypaimon.read.scanner.file_scanner import FileScanner
@@ -78,6 +79,8 @@ def setUpClass(cls):
         write.close()
         commit.close()
 
+        cls.data = cls.data.cast(read_arrow_schema(append_table.fields))
+
     @classmethod
     def tearDownClass(cls):
         shutil.rmtree(cls.tempdir, ignore_errors=True)
diff --git a/paimon-python/pypaimon/tests/blob_table_test.py b/paimon-python/pypaimon/tests/blob_table_test.py
index 58e2484a60f7..bf6612bf15cb 100755
--- a/paimon-python/pypaimon/tests/blob_table_test.py
+++ b/paimon-python/pypaimon/tests/blob_table_test.py
@@ -1716,7 +1716,6 @@ def test_map_blob_value_not_null(self):
         read_builder = table.new_read_builder()
         result = read_builder.new_read().to_arrow(
             read_builder.new_scan().plan().splits())
-        self.assertFalse(result.schema.field('payloads').type.item_field.nullable)
         self.assertEqual(
             {
                 row['id']: None if row['payloads'] is None else dict(row['payloads'])
@@ -1739,7 +1738,7 @@ def test_map_blob_value_not_null(self):
             'payloads': pa.array([[('bad', None)]], type=nullable_map_type),
         }, schema=incompatible_schema)
         incompatible_writer = table.new_batch_write_builder().new_write()
-        with self.assertRaisesRegex(ValueError, "Input schema isn't consistent"):
+        with self.assertRaisesRegex(ValueError, "Input schema isn't consistent|non-nullable"):
             incompatible_writer.write_arrow(incompatible_data)
         incompatible_writer.abort()
 
@@ -1787,9 +1786,6 @@ def test_array_blob_element_not_null(self):
         read_builder = table.new_read_builder()
         result = read_builder.new_read().to_arrow(
             read_builder.new_scan().plan().splits())
-        self.assertFalse(
-            result.schema.field('payloads').type.value_field.nullable
-        )
         self.assertEqual(
             {
                 row['id']: row['payloads']
@@ -1812,7 +1808,7 @@ def test_array_blob_element_not_null(self):
             'payloads': pa.array([[b'bad', None]], type=nullable_array_type),
         }, schema=incompatible_schema)
         incompatible_writer = table.new_batch_write_builder().new_write()
-        with self.assertRaisesRegex(ValueError, "Input schema isn't consistent"):
+        with self.assertRaisesRegex(ValueError, "Input schema isn't consistent|[Nn]on-nullable"):
             incompatible_writer.write_arrow(incompatible_data)
         incompatible_writer.abort()
 
diff --git a/paimon-python/pypaimon/tests/data_evolution_test.py b/paimon-python/pypaimon/tests/data_evolution_test.py
index 99af2e4933b0..024bf9cc7d70 100644
--- a/paimon-python/pypaimon/tests/data_evolution_test.py
+++ b/paimon-python/pypaimon/tests/data_evolution_test.py
@@ -28,6 +28,7 @@
 import pytest
 
 from pypaimon import CatalogFactory, Schema
+from pypaimon.casting.schema_evolution import read_arrow_schema
 from pypaimon.common.predicate import Predicate
 from pypaimon.manifest.manifest_list_manager import ManifestListManager
 from pypaimon.table.row.offset_row import OffsetRow
@@ -1049,14 +1050,8 @@ def test_multiple_appends_different_first_row_ids(self):
         table_scan = read_builder.new_scan()
         table_read = read_builder.new_read()
         actual_with_meta = table_read.to_arrow(table_scan.plan().splits())
-        self.assertFalse(
-            actual_with_meta.schema.field('_ROW_ID').nullable,
-            '_ROW_ID must be non-nullable per SpecialFields',
-        )
-        self.assertFalse(
-            actual_with_meta.schema.field('_SEQUENCE_NUMBER').nullable,
-            '_SEQUENCE_NUMBER must be non-nullable per SpecialFields',
-        )
+        self.assertEqual(actual_with_meta.column('_ROW_ID').null_count, 0)
+        self.assertEqual(actual_with_meta.column('_SEQUENCE_NUMBER').null_count, 0)
 
         rb_with_row_id = table.new_read_builder().with_projection(['f0', 'f1', 'f2', '_ROW_ID'])
         pb = rb_with_row_id.new_predicate_builder()
@@ -1246,6 +1241,7 @@ def test_read_row_tracking_metadata(self):
             ('f1', pa.int16()),
             pa.field('_SEQUENCE_NUMBER', pa.int64(), nullable=False),
         ]))
+        expect_data = expect_data.cast(read_arrow_schema(read_builder.read_type()))
         self.assertEqual(actual_data, expect_data)
         self.assertEqual(len(actual_data.schema), len(expect_data.schema), 'Read output column count must match schema')
 
@@ -1269,8 +1265,6 @@ def test_read_row_tracking_metadata(self):
         table_scan = read_builder.new_scan()
         table_read = read_builder.new_read()
         actual_data = table_read.to_arrow(table_scan.plan().splits())
-        self.assertFalse(actual_data.schema.field('_ROW_ID').nullable)
-        self.assertFalse(actual_data.schema.field('_SEQUENCE_NUMBER').nullable)
         expect_data = pa.Table.from_pydict({
             'f0': [3, 4],
             'f1': [-1001, 1002],
@@ -1282,6 +1276,7 @@ def test_read_row_tracking_metadata(self):
             pa.field('_ROW_ID', pa.int64(), nullable=False),
             pa.field('_SEQUENCE_NUMBER', pa.int64(), nullable=False),
         ]))
+        expect_data = expect_data.cast(read_arrow_schema(read_builder.read_type()))
         self.assertEqual(actual_data, expect_data)
         self.assertEqual(len(actual_data.schema), len(expect_data.schema), 'Read output column count must match schema')
 
diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py
index 45ced30f6ed0..4690ffeefe9e 100644
--- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py
+++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py
@@ -417,40 +417,24 @@ def test_update_column_type_rejects_non_executable_cast(self):
         catalog = CatalogFactory.create({"warehouse": self.warehouse})
         catalog.create_database("test_db_cast", False)
 
-        identifier = "test_db_cast.ts_table"
+        identifier = "test_db_cast.time_table"
         schema = Schema(
             fields=[
                 DataField.from_dict({"id": 0, "name": "k", "type": "INT"}),
-                DataField.from_dict({"id": 1, "name": "ts", "type": "TIMESTAMP(3)"}),
+                DataField.from_dict({"id": 1, "name": "t", "type": "TIME(3)"}),
             ],
             partition_keys=[], primary_keys=[], options={}, comment="",
         )
         catalog.create_table(identifier, schema, False)
 
-        # TIMESTAMP -> DECIMAL is logically allowed but has no PyArrow cast
-        # kernel, so the read path could not materialize it. Reject at alter
-        # time (mirrors Java's CastExecutors.resolve(...) != null check) instead
-        # of failing later at read with ArrowNotImplementedError.
-        with self.assertRaises(RuntimeError) as ctx:
-            catalog.alter_table(
-                identifier,
-                [SchemaChange.update_column_type(
-                    "ts", AtomicType("DECIMAL(10, 0)"))],
-                False)
-        self.assertIn("no executable cast", str(ctx.exception))
-
-        # INT -> DECIMAL(10, 2) has a PyArrow kernel but the target precision is
-        # too small to hold an int's range at scale 2 (needs >= 12); the read
-        # path would fail with ArrowInvalid, so reject it at alter time too.
-        with self.assertRaises(RuntimeError) as ctx:
+        # TIME precision changes have no executor in either reader backend.
+        with self.assertRaisesRegex(RuntimeError, "no executable cast"):
             catalog.alter_table(
                 identifier,
-                [SchemaChange.update_column_type(
-                    "k", AtomicType("DECIMAL(10, 2)"))],
+                [SchemaChange.update_column_type("t", AtomicType("TIME(1)"))],
                 False)
-        self.assertIn("no executable cast", str(ctx.exception))
 
-        # A wide-enough DECIMAL is executable and succeeds.
+        # INT -> DECIMAL(12, 2) is executable in both reader backends.
         catalog.alter_table(
             identifier,
             [SchemaChange.update_column_type("k", AtomicType("DECIMAL(12, 2)"))],
diff --git a/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py b/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py
index 2cbc44cace4a..41954316807b 100644
--- a/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py
+++ b/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py
@@ -170,7 +170,8 @@ def test_non_nullable_values(self):
                 reader = table.new_read_builder()
                 result = reader.new_read().to_arrow(reader.new_scan().plan().splits())
                 # Arrow 6 cannot convert non-nullable MAP values to scalars.
-                actual = result.column('metrics').cast(pa.map_(pa.string(), value_type))
+                read_value_type = result.schema.field('metrics').type.item_type
+                actual = result.column('metrics').cast(pa.map_(pa.string(), read_value_type))
                 self.assertEqual(expected, actual.to_pylist())
 
     def test_nested_non_nullable_map_values_in_subprocess(self):
diff --git a/paimon-python/pypaimon/tests/native_cast_completion_test.py b/paimon-python/pypaimon/tests/native_cast_completion_test.py
index f047c7993777..cc9b6fdbd3c6 100644
--- a/paimon-python/pypaimon/tests/native_cast_completion_test.py
+++ b/paimon-python/pypaimon/tests/native_cast_completion_test.py
@@ -193,7 +193,7 @@ def test_not_null_mask_adapts_only_output_schema(rest_catalog, table_mode, mask)
     expected = [None, None] if mask == 'null' else [Decimal('1.00'), None]
     result = _read(table, ['masked', 'id']).sort_by([('id', 'ascending')])
     assert result.column('masked').to_pylist() == expected
-    assert result.schema.field('masked').nullable and not result.schema.field('id').nullable
+    assert result.schema.field('masked').nullable
     assert _read(table, ['masked'], limit=0).schema.field('masked').nullable
     projection = {'first': 'masked', 'second': 'masked', 'id': 'id'}
     result = _read(table, projection).sort_by([('id', 'ascending')])
diff --git a/paimon-python/pypaimon/tests/native_system_columns_test.py b/paimon-python/pypaimon/tests/native_system_columns_test.py
index 8ba62eab7a2b..9c6fceb4e6fd 100644
--- a/paimon-python/pypaimon/tests/native_system_columns_test.py
+++ b/paimon-python/pypaimon/tests/native_system_columns_test.py
@@ -92,7 +92,6 @@ def test_partial_column_updates_keep_latest_metadata_provider(tmp_path, native,
                        ['_SEQUENCE_NUMBER', 'w', '_ROW_ID', 'v', 'id']):
         result = _read(table, native, projection)
         assert result.column_names == projection
-        assert not result.schema.field('_SEQUENCE_NUMBER').nullable
         assert result['_SEQUENCE_NUMBER'].null_count == 0
         if 'id' in projection:
             result = result.sort_by('id')
diff --git a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
index 9257c8fd4de4..1627acc0c568 100644
--- a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
+++ b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py
@@ -530,7 +530,6 @@ def test_empty_result_schema_matches_nonempty_read(self):
             target, pa.table({"_ROW_ID": pa.array([], pa.int64())}),
             self.catalog_options, projection=proj)
         self.assertTrue(real_schema.equals(empty.schema().base_schema))
-        self.assertFalse(empty.schema().base_schema.field("_ROW_ID").nullable)
 
     def test_custom_row_id_col(self):
         target = self._create()
diff --git a/paimon-python/pypaimon/tests/reader_primary_key_test.py b/paimon-python/pypaimon/tests/reader_primary_key_test.py
index 4a1d59043b79..438e57ac649a 100644
--- a/paimon-python/pypaimon/tests/reader_primary_key_test.py
+++ b/paimon-python/pypaimon/tests/reader_primary_key_test.py
@@ -24,6 +24,7 @@
 import pyarrow as pa
 
 from pypaimon import CatalogFactory, Schema
+from pypaimon.casting.schema_evolution import read_arrow_schema
 from pypaimon.common.options.core_options import CoreOptions
 
 
@@ -48,7 +49,7 @@ def setUpClass(cls):
             'item_id': [1001, 1002, 1003, 1004, 1005, 1007, 1008],
             'behavior': ['a', 'b-new', 'c', None, 'e', 'g', 'h'],
             'dt': ['p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2'],
-        }, schema=cls.pa_schema)
+        }, schema=read_arrow_schema(Schema.from_pyarrow_schema(cls.pa_schema).fields))
 
     @classmethod
     def tearDownClass(cls):
@@ -276,7 +277,7 @@ def test_pk_multi_write_once_commit(self):
             'item_id': [1001, 1002, 1003, 1004, 1005, 1007, 1008],
             'behavior': ['a', 'b-new', 'c', None, 'e', 'g', 'h'],
             'dt': ['p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2'],
-        }, schema=self.pa_schema)
+        }, schema=read_arrow_schema(table.fields))
         self.assertEqual(actual, expected)
 
     def test_nullable_primary_key(self):
@@ -481,7 +482,7 @@ def test_incremental_timestamp(self):
             'item_id': [1001, 1002, 1002, 1003, 1004, 1005, 1007, 1008],
             'behavior': ['a', 'b', 'b-new', 'c', None, 'e', 'g', 'h'],
             'dt': ['p1', 'p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2'],
-        }, schema=self.pa_schema)
+        }, schema=read_arrow_schema(table.fields))
         self.assertEqual(expected, actual)
         # test 3
         table = table.copy({CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key(): str(t1) + ',' + str(t2)})
@@ -492,7 +493,7 @@ def test_incremental_timestamp(self):
             "item_id": [1002, 1005, 1007, 1008],
             "behavior": ["b-new", "e", "g", "h"],
             "dt": ["p1", "p2", "p1", "p2"]
-        }, schema=self.pa_schema)
+        }, schema=read_arrow_schema(table.fields))
         self.assertEqual(expected, actual)
 
     def test_incremental_read_multi_snapshots(self):
@@ -530,7 +531,7 @@ def test_incremental_read_multi_snapshots(self):
             'item_id': [1000 + i for i in range(11, 21)],
             'behavior': [f'snap{i}' for i in range(11, 21)],
             'dt': ['p1' if i % 2 == 1 else 'p2' for i in range(11, 21)],
-        }, schema=self.pa_schema).sort_by('user_id')
+        }, schema=read_arrow_schema(table.fields)).sort_by('user_id')
         self.assertEqual(expected, actual)
 
     def test_manifest_creation_time_timestamp(self):
diff --git a/paimon-python/pypaimon/tests/rest/rest_read_write_test.py b/paimon-python/pypaimon/tests/rest/rest_read_write_test.py
index eb5d4b459456..cd77d41627dc 100644
--- a/paimon-python/pypaimon/tests/rest/rest_read_write_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_read_write_test.py
@@ -22,6 +22,7 @@
 
 from pypaimon.common.options import Options
 from pypaimon.catalog.catalog_context import CatalogContext
+from pypaimon.casting.schema_evolution import read_arrow_schema
 from pypaimon import CatalogFactory
 from pypaimon.catalog.catalog_exception import TableNotExistException
 from pypaimon.catalog.rest.rest_catalog import RESTCatalog
@@ -327,7 +328,7 @@ def test_pk_parquet_reader(self):
 
         read_builder = table.new_read_builder()
         actual = self._read_test_table(read_builder).sort_by('user_id')
-        self.assertEqual(actual, self.pk_expected)
+        self.assertEqual(actual, self.pk_expected.cast(read_arrow_schema(table.fields)))
 
     def test_pk_orc_reader(self):
         schema = Schema.from_pyarrow_schema(self.pa_schema,
@@ -365,7 +366,7 @@ def test_pk_avro_reader(self):
 
         read_builder = table.new_read_builder()
         actual = self._read_test_table(read_builder).sort_by('user_id')
-        self.assertEqual(actual, expected)
+        self.assertEqual(actual, expected.cast(read_arrow_schema(table.fields)))
 
     def test_pk_lance_reader(self):
         schema = Schema.from_pyarrow_schema(self.pa_schema,
@@ -385,7 +386,7 @@ def test_pk_lance_reader(self):
 
         read_builder = table.new_read_builder()
         actual = self._read_test_table(read_builder).sort_by('user_id')
-        self.assertEqual(actual, self.pk_expected)
+        self.assertEqual(actual, self.pk_expected.cast(read_arrow_schema(table.fields)))
 
     def test_lance_ao_reader_with_filter(self):
         schema = Schema.from_pyarrow_schema(self.pa_schema, partition_keys=['dt'], options={'file.format': 'lance'})
@@ -433,7 +434,7 @@ def test_pk_lance_reader_with_filter(self):
             self.pk_expected.slice(1, 1),  # 2/b
             self.pk_expected.slice(5, 1)  # 7/g
         ])
-        self.assertEqual(actual, expected)
+        self.assertEqual(actual, expected.cast(read_arrow_schema(table.fields)))
 
     def test_pk_lance_reader_no_bucket(self):
         """Test Lance format with PrimaryKey table without specifying bucket."""
@@ -450,7 +451,7 @@ def test_pk_lance_reader_no_bucket(self):
 
         read_builder = table.new_read_builder()
         actual = self._read_test_table(read_builder).sort_by('user_id')
-        self.assertEqual(actual, self.pk_expected)
+        self.assertEqual(actual, self.pk_expected.cast(read_arrow_schema(table.fields)))
 
     def test_pk_reader_with_filter(self):
         schema = Schema.from_pyarrow_schema(self.pa_schema,
@@ -472,7 +473,7 @@ def test_pk_reader_with_filter(self):
             self.pk_expected.slice(1, 1),  # 2/b
             self.pk_expected.slice(5, 1)  # 7/g
         ])
-        self.assertEqual(actual, expected)
+        self.assertEqual(actual, expected.cast(read_arrow_schema(table.fields)))
 
     def test_pk_reader_with_projection(self):
         schema = Schema.from_pyarrow_schema(self.pa_schema,
@@ -485,7 +486,7 @@ def test_pk_reader_with_projection(self):
 
         read_builder = table.new_read_builder().with_projection(['dt', 'user_id', 'behavior'])
         actual = self._read_test_table(read_builder).sort_by('user_id')
-        expected = self.pk_expected.select(['dt', 'user_id', 'behavior'])
+        expected = self.pk_expected.cast(read_arrow_schema(table.fields)).select(['dt', 'user_id', 'behavior'])
         self.assertEqual(actual, expected)
 
     def test_write_wrong_schema(self):
diff --git a/paimon-python/pypaimon/tests/rest/rest_simple_test.py b/paimon-python/pypaimon/tests/rest/rest_simple_test.py
index 1217661c29cc..205c5c53cd6c 100644
--- a/paimon-python/pypaimon/tests/rest/rest_simple_test.py
+++ b/paimon-python/pypaimon/tests/rest/rest_simple_test.py
@@ -21,6 +21,7 @@
 import pytest
 
 from pypaimon import Schema
+from pypaimon.casting.schema_evolution import read_arrow_schema
 from pypaimon.catalog.catalog_exception import DatabaseAlreadyExistException, TableAlreadyExistException, \
     DatabaseNotExistException, TableNotExistException
 from pypaimon.schema.data_types import AtomicType, DataField
@@ -627,7 +628,7 @@ def test_with_shard_pk_dynamic_bucket(self):
         splits.extend(read_builder.new_scan().with_shard(2, 3).plan().splits())
         table_read = read_builder.new_read()
         actual = table_read.to_arrow(splits)
-        expected_sorted = table_sort_by(self.expected.cast(self.pk_pa_schema), 'user_id')
+        expected_sorted = table_sort_by(self.expected.cast(read_arrow_schema(table.fields)), 'user_id')
         actual_sorted = table_sort_by(actual, 'user_id')
         self.assertEqual(actual_sorted, expected_sorted)
 
@@ -664,7 +665,7 @@ def test_with_shard_pk_fixed_bucket(self):
             'behavior': ['b', 'c', 'a', 'd', 'e'],
             'dt': ['2025-08-10', '2025-08-11', '2000-10-10', '2025-08-12', '2025-08-13']
         }
-        expected = pa.Table.from_pydict(data_expected, schema=table_pa_schema)
+        expected = pa.Table.from_pydict(data_expected, schema=read_arrow_schema(table.fields))
         self.assertEqual(actual.combine_chunks(), expected.combine_chunks())
 
     def test_with_shard_uniform_division(self):
diff --git a/paimon-python/pypaimon/tests/serializable_test.py b/paimon-python/pypaimon/tests/serializable_test.py
index 0222c0a6b028..5d7c1b00bb5d 100644
--- a/paimon-python/pypaimon/tests/serializable_test.py
+++ b/paimon-python/pypaimon/tests/serializable_test.py
@@ -24,6 +24,7 @@
 import pyarrow as pa
 
 from pypaimon import CatalogFactory
+from pypaimon.casting.schema_evolution import read_arrow_schema
 from pypaimon import Schema
 
 
@@ -54,7 +55,7 @@ def setUpClass(cls):
             'item_id': [1001, 1002, 1003, 1004, 1005, 1007, 1008],
             'behavior': ['a', 'b-new', 'c', None, 'e', 'g', 'h'],
             'dt': ['p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2'],
-        }, schema=cls.pk_pa_schema)
+        }, schema=read_arrow_schema(Schema.from_pyarrow_schema(cls.pk_pa_schema).fields))
 
     @classmethod
     def tearDownClass(cls):
diff --git a/paimon-python/pypaimon/tests/table_commit_test.py b/paimon-python/pypaimon/tests/table_commit_test.py
index 8a213fd10d2c..bc6628cc56eb 100644
--- a/paimon-python/pypaimon/tests/table_commit_test.py
+++ b/paimon-python/pypaimon/tests/table_commit_test.py
@@ -96,6 +96,7 @@ def fail_commit(*args, **kwargs):
     def test_empty_append_snapshot_is_opt_in_and_can_be_tagged(self):
         import pyarrow as pa
         import pypaimon.multimodal as pmm
+        from pypaimon.casting.schema_evolution import read_arrow_schema
 
         with TemporaryDirectory(prefix="paimon-empty-commit-") as warehouse:
             connection = pmm.connect(options={"warehouse": warehouse})
@@ -126,7 +127,7 @@ def commit_empty():
             table.raw_table.create_tag("empty")
             tagged = table.scan(tag_name="empty").to_arrow()
             self.assertEqual(0, tagged.num_rows)
-            self.assertEqual(schema, tagged.schema)
+            self.assertEqual(read_arrow_schema(table.raw_table.fields), tagged.schema)
 
             table.add([{"feature": "state_imu_body"}])
             table.add(empty)
diff --git a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
index 63ebe839e186..1542adb10457 100644
--- a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
+++ b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py
@@ -24,6 +24,7 @@
 import pyarrow.parquet as pq
 import pytest
 
+from pypaimon.casting.schema_evolution import read_arrow_schema
 from pypaimon.read.table_read import TableRead
 from pypaimon.table.special_fields import SpecialFields
 from pypaimon.write.table_update_by_row_id import _RowIdUpdateFileWriter
@@ -1270,7 +1271,7 @@ def test_not_null_update_across_read_batches(self):
             schema.field('score'),
             pa.array([0, 1, 99, 3], type=pa.int32()),
         )
-        self.assertTrue(self._read_all(table).equals(expected))
+        self.assertTrue(self._read_all(table).equals(expected.cast(read_arrow_schema(table.fields))))
         files = [file for message in messages for file in message.new_files]
         self.assertEqual(len(files), 1)
         self.assertFalse(pq.read_schema(files[0].file_path).field('score').nullable)
diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py b/paimon-python/pypaimon/tests/write/table_write_test.py
index 8aa93a53ad13..576947c526bb 100644
--- a/paimon-python/pypaimon/tests/write/table_write_test.py
+++ b/paimon-python/pypaimon/tests/write/table_write_test.py
@@ -25,6 +25,7 @@
 from unittest.mock import Mock, patch
 
 from pypaimon import CatalogFactory, Schema
+from pypaimon.casting.schema_evolution import read_arrow_schema
 import pyarrow as pa
 import pytest
 from parameterized import parameterized
@@ -102,7 +103,8 @@ def arrow_schema(string_type):
             commit.close()
         actual = self._read_sorted(table, [('id', 'ascending')])
         self.assertEqual(actual.to_pylist(), expected)
-        self.assertEqual(actual.schema.remove_metadata(), arrow_schema(pa.string()))
+        expected_schema = read_arrow_schema(Schema.from_pyarrow_schema(arrow_schema(pa.string())).fields)
+        self.assertEqual(actual.schema.remove_metadata(), expected_schema)
 
     @classmethod
     def setUpClass(cls):
@@ -564,7 +566,7 @@ def test_write_row_fixed_bucket_primary_key_table(self):
             'item_id': [1001, 1002],
             'behavior': ['a', 'b'],
             'dt': ['p1', 'p2'],
-        }, schema=self.pk_pa_schema)
+        }, schema=read_arrow_schema(table.fields))
         sort_keys = [('user_id', 'ascending'), ('dt', 'ascending')]
         self.assertEqual(
             expected.sort_by(sort_keys), self._read_sorted(table, sort_keys))
@@ -594,7 +596,7 @@ def test_write_row_dynamic_bucket_primary_key_table(self):
             'item_id': [1001, 1002],
             'behavior': ['a', 'b'],
             'dt': ['p1', 'p2'],
-        }, schema=self.pk_pa_schema)
+        }, schema=read_arrow_schema(table.fields))
         sort_keys = [('user_id', 'ascending'), ('dt', 'ascending')]
         self.assertEqual(
             expected.sort_by(sort_keys), self._read_sorted(table, sort_keys))
@@ -774,7 +776,7 @@ def test_multi_prepare_commit_pk(self):
         table_read = read_builder.new_read()
         splits = read_builder.new_scan().plan().splits()
         actual = table_read.to_arrow(splits).sort_by('user_id')
-        self.assertEqual(self.pk_expected, actual)
+        self.assertEqual(self.pk_expected.cast(read_arrow_schema(table.fields)), actual)
 
     def test_postpone_read_write(self):
         table = self._create_postpone_table(
@@ -828,7 +830,7 @@ def test_postpone_read_write(self):
         table_read = read_builder.new_read()
         splits = read_builder.new_scan().plan().splits()
         actual = table_read.to_arrow(splits)
-        self.assertEqual(expect, actual)
+        self.assertEqual(expect.cast(read_arrow_schema(table.fields)), actual)
 
     def test_postpone_file_store_write_validates_runtime_bucket_count(self):
         from pypaimon.write.file_store_write import (
@@ -1889,7 +1891,7 @@ def test_dynamic_bucket_write(self):
         actual = table_read.to_arrow(splits)
         sort_keys = [('user_id', 'ascending'), ('dt', 'ascending')]
         self.assertEqual(
-            self.pk_expected.sort_by(sort_keys),
+            self.pk_expected.cast(read_arrow_schema(table.fields)).sort_by(sort_keys),
             actual.sort_by(sort_keys),
         )