[python] Align empty and metadata-only results with runtime read schemas

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:

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(

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)

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()
 

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')
 

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)"))],

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):

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')])

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')

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()

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):

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):

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):

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):