[python] Delegate schema evolution and runtime read schemas to Rust core

.github/workflows/ci-python.yml

@@ -214,6 +214,12 @@ jobs:
           source = json.loads(distribution.read_text('direct_url.json'))
           print('pypaimon-rust', distribution.version,
                 'commit', source['vcs_info']['commit_id'])
+          # Fail before the suite when Rust main lacks the required core bridge.
+          from pypaimon_rust.data import (
+              cast_array_for_schema_evolution,
+              normalize_array_for_write,
+              read_arrow_schema,
+          )
           PY
 
       - name: Run Python tests with native planning, reading, writing, updating and committing
@@ -239,7 +245,8 @@ jobs:
             TZ="$zone" python -m pytest -q \
               paimon-python/pypaimon/tests/native_cast_completion_test.py \
               paimon-python/pypaimon/tests/native_temporal_zone_cast_test.py \
-              -k 'TIMESTAMP_LTZ or temporal'
+              paimon-python/pypaimon/tests/native_schema_cast_test.py \
+              -k 'TIMESTAMP or temporal'
           done
 
   torch:

paimon-python/pypaimon/casting/data_type_casts.py

@@ -216,6 +216,11 @@ def can_execute_cast(source_type, target_type) -> bool:
     # Same root: identity, or a same-shape constructed type whose value is
     # rebuilt by the read path's field-id alignment rather than a value cast.
     if source_root == target_root:
+        if source_root == TIME:
+            from pypaimon.table.row.generic_row import _parse_type_precision_scale
+            # Compare parsed precision: TIME and TIME(0) describe the same type.
+            return (_parse_type_precision_scale(source_type.type)[0]
+                    == _parse_type_precision_scale(target_type.type)[0])
         return True
     # Constructed -> character string is rendered by the read path's custom
     # ``_constructed_to_string_array`` (see DataFileBatchReader), not a cast.
@@ -254,7 +259,7 @@ def _schema_evolution_cast_supported(source_type, target_type) -> bool:
             pa.nulls(1, type=source_pa), target_pa, source_type, target_type)
         ok = True
     except (pa.lib.ArrowNotImplementedError, pa.lib.ArrowInvalid,
-            pa.lib.ArrowTypeError):
+            pa.lib.ArrowTypeError, ValueError, NotImplementedError):
         ok = False
     _EXECUTABLE_CAST_CACHE[cache_key] = ok
     return ok

paimon-python/pypaimon/casting/schema_evolution.py

@@ -18,6 +18,7 @@
 """Java cast rules shared by old-file readers."""
 
 import math
+import json
 
 import pyarrow as pa
 
@@ -69,9 +70,16 @@ def _truncate_timestamp(array, target_type, source_paimon_type, target_paimon_ty
         return None
     nanos_per_tick = {'s': 1000000000, 'ms': 1000000, 'us': 1000, 'ns': 1}
     quantum = 10 ** (9 - target_precision)
-    values = [None if value is None else
-              (value * nanos_per_tick[array.type.unit] // quantum * quantum) // nanos_per_tick[target_type.unit]
-              for value in array.cast(pa.int64()).to_pylist()]
+    values = []
+    for value in array.cast(pa.int64()).to_pylist():
+        if value is None:
+            values.append(None)
+            continue
+        nanos = value * nanos_per_tick[array.type.unit] // quantum * quantum
+        millis, remainder = divmod(nanos, 1000000)
+        # Java stores signed-long milliseconds before converting storage units.
+        millis = _narrow_integer(millis, 64)
+        values.append((millis * 1000000 + remainder) // nanos_per_tick[target_type.unit])
     # Flooring a nanosecond boundary can exceed Arrow's int64 storage even
     # though Java's timestamp representation can hold it. Match Native's
     # lenient overflow handling when rebuilding the Arrow result.
@@ -114,6 +122,33 @@ def _integer_to_timestamp(array, target_type):
     return pa.array(values, type=pa.int64()).cast(target_type)
 
 
+def _native_schema_cast(array, target_type, source_paimon_type, target_paimon_type):
+    if source_paimon_type is None or target_paimon_type is None:
+        return None
+    try:
+        from pypaimon_rust.data import cast_array_for_schema_evolution as native_cast
+    except ImportError:
+        return None
+    source = json.dumps(source_paimon_type.to_dict())
+    target = json.dumps(target_paimon_type.to_dict())
+    if isinstance(array, pa.ChunkedArray):
+        chunks = array.chunks or [pa.array([], type=array.type)]
+        # Runtime cast results may be NULL despite a logical NOT NULL child.
+        return pa.chunked_array([native_cast(chunk, source, target) for chunk in chunks])
+    return native_cast(array, source, target)
+
+
+def read_arrow_schema(fields):
+    """Use core runtime nullability while retaining logical write constraints."""
+    try:
+        from pypaimon_rust.data import read_arrow_schema as native_schema
+    except ImportError:
+        from pypaimon.schema.data_types import PyarrowFieldParser
+        return PyarrowFieldParser.from_paimon_schema(fields)
+    schema = native_schema(json.dumps([field.to_dict() for field in fields]))
+    return pa.schema([field.remove_metadata() for field in schema])
+
+
 def cast_array_for_schema_evolution(array, target_type, source_paimon_type=None, target_paimon_type=None):
     """Materialize existing type changes using Java numeric and timestamp casts.
 
@@ -126,6 +161,9 @@ def cast_array_for_schema_evolution(array, target_type, source_paimon_type=None,
     precision overflow after rounding.
     LTZ conversions use the process local zone, including Java gap/overlap rules.
     """
+    native = _native_schema_cast(array, target_type, source_paimon_type, target_paimon_type)
+    if native is not None:
+        return native
     if pa.types.is_string(target_type):
         strings = temporal_to_string(array, source_paimon_type, target_paimon_type)
         if strings is not None:

paimon-python/pypaimon/read/reader/data_file_batch_reader.py

@@ -21,7 +21,7 @@
 import pyarrow.compute as pc
 from pyarrow import RecordBatch
 
-from pypaimon.casting.schema_evolution import cast_array_for_schema_evolution
+from pypaimon.casting.schema_evolution import cast_array_for_schema_evolution, read_arrow_schema
 from pypaimon.common.file_io import FileIO
 from pypaimon.utils.arrow_utils import zero_column_batch
 from pypaimon.data.map_shared_shredding import (
@@ -68,6 +68,10 @@ def _constructed_to_string_array(array, file_type):
     ROW -> ``{v1, v2}``, ARRAY -> ``[e1, e2]``, MAP -> ``{k1 -> v1, k2 -> v2}``.
     Sub-values are rendered recursively; a NULL sub-value renders as the
     literal ``null`` while a NULL container row stays NULL."""
+    from pypaimon.casting.schema_evolution import _native_schema_cast
+    native = _native_schema_cast(array, pa.string(), file_type, AtomicType('STRING'))
+    if native is not None:
+        return native
     array = _unslice(array)
     valid = pc.is_valid(array).to_pylist()
     out = []
@@ -172,7 +176,7 @@ def __init__(self, format_reader: RecordBatchReader, index_mapping: List[int], p
         self._file_data_field_map = {
             field.name: field for field in source_fields
         }
-        self.schema_map = {field.name: field for field in PyarrowFieldParser.from_paimon_schema(fields)}
+        self.schema_map = {field.name: field for field in read_arrow_schema(fields)}
         self.row_tracking_enabled = row_tracking_enabled
         self.first_row_id = first_row_id
         self.row_id_offsets = row_id_offsets
@@ -264,6 +268,11 @@ def _align_array_by_id(self, array, file_type, target_type):
         """Return *array* converted to *target_type*, matching ROW sub-fields by
         field id (reorder, pad missing with NULL, follow renames, cast changed
         types) recursively, transparently through ARRAY/MAP wrappers."""
+        from pypaimon.casting.schema_evolution import _native_schema_cast
+        native = _native_schema_cast(
+            array, PyarrowFieldParser.from_paimon_type(target_type), file_type, target_type)
+        if native is not None:
+            return native
         if isinstance(target_type, RowType) and isinstance(file_type, RowType):
             n = len(array)
             file_id_to_pos = {f.id: i for i, f in enumerate(file_type.fields)}

paimon-python/pypaimon/read/table_read.py

@@ -25,6 +25,8 @@
 import pandas
 import pyarrow
 
+from pypaimon.casting.schema_evolution import read_arrow_schema
+
 from pypaimon.common.options.core_options import CoreOptions
 from pypaimon.common.predicate import Predicate
 from pypaimon.common.predicate_json_parser import extract_referenced_fields
@@ -46,7 +48,7 @@
                                       MergeFileSplitRead, RawFileSplitRead,
                                       SplitRead, deferred_blob_field_names)
 from pypaimon.schema.data_types import (
-    DataField, MapType, PyarrowFieldParser, is_blob_file_field, is_map_blob_type)
+    DataField, MapType, is_blob_file_field, is_map_blob_type)
 from pypaimon.table.row.offset_row import OffsetRow
 from pypaimon.utils.arrow_utils import zero_column_batch
 
@@ -464,7 +466,7 @@ def _reader_adapter(self):
         return projected if predicate_field_names(self.predicate) & extra_names else adapter
 
     def _output_arrow_schema(self, splits=None):
-        schema = PyarrowFieldParser.from_paimon_schema(self.read_type)
+        schema = read_arrow_schema(self.read_type)
         masks = set()
         if splits:
             for split in splits:
@@ -851,7 +853,12 @@ def _prepare_native_batch(self, batch, schema):
     def _flatten_native_nested_batch(self, batch, schema):
         _, paths = self._reader_adapter()
         arrays = [extract_array(batch, path) for path in paths]
-        fields = list(PyarrowFieldParser.from_paimon_schema(self._adapter_read_type))
+        fields = list(read_arrow_schema(self._adapter_read_type))
+        # Authorization can mask a NOT NULL field. Keep the captured source
+        # nullability when flattening, including whole-parent/MAP projections.
+        projected = output_schema(batch.schema, OutputProjection([
+            (field.name, path) for field, path in zip(fields, paths)]))
+        fields = [field.with_nullable(field.nullable or projected[i].nullable) for i, field in enumerate(fields)]
         if self.include_row_kind:
             arrays.insert(0, batch.column(ROW_KIND_COLUMN))
             fields.insert(0, batch.schema.field(ROW_KIND_COLUMN))
@@ -1159,12 +1166,13 @@ def _convert_rows_to_output_batches(
             yield from self._convert_rows_to_arrow_batches_with_row_kind(
                 row_tuples, row_kinds, schema)
             return
-        physical_schema = PyarrowFieldParser.from_paimon_schema(self._adapter_read_type)
+        physical_schema = read_arrow_schema(self._adapter_read_type)
         if self.include_row_kind:
             physical_schema = self._add_row_kind_to_schema(physical_schema)
         for batch in self._convert_rows_to_arrow_batches_with_row_kind(
                 row_tuples, row_kinds, physical_schema):
-            yield self._project_batch_to_output(batch)
+            yield self._try_to_pad_batch_by_schema(
+                self._project_batch_to_output(batch), schema, allow_type_cast=True)
 
     def _convert_rows_to_arrow_batches_with_row_kind(
         self,
@@ -1665,8 +1673,8 @@ def _project_batch_to_output(self, batch: pyarrow.RecordBatch) -> pyarrow.Record
             # Project the already masked schema, retaining captured nullability.
             adapted_projection = OutputProjection([(alias, path) for alias, path in outputs])
             actual = output_schema(batch.schema, adapted_projection)
-            declared = output_schema(PyarrowFieldParser.from_paimon_schema(self.read_type), self.output_projection)
-            fields = [field.with_nullable(actual[i].nullable) for i, field in enumerate(declared)]
+            declared = output_schema(read_arrow_schema(self.read_type), self.output_projection)
+            fields = [field.with_nullable(field.nullable or actual[i].nullable) for i, field in enumerate(declared)]
             if self.include_row_kind and ROW_KIND_COLUMN in batch.schema.names:
                 arrays.insert(0, batch.column(ROW_KIND_COLUMN))
                 fields.insert(0, batch.schema.field(ROW_KIND_COLUMN))
@@ -1873,7 +1881,7 @@ def __authed_reader(self, split, auth_result, blob_parallelism=1,
 
         needs_convert_back = False
         if not isinstance(reader, RecordBatchReader):
-            schema = PyarrowFieldParser.from_paimon_schema(effective_read_type)
+            schema = read_arrow_schema(effective_read_type)
             reader = RecordReaderToBatchAdapter(reader, schema, include_row_kind=self.include_row_kind)
             if getattr(reader, 'blob_field_indices', None) is None:
                 from pypaimon.read.reader.field_indices import (

paimon-python/pypaimon/schema/arrow_schema.py

@@ -20,14 +20,18 @@
 import pyarrow as pa
 
 
-def _normalize_string_type(data_type):
+def _normalize_string_type(data_type, ignore_child_nullability=False):
+    def child(field):
+        field = field.with_type(_normalize_string_type(field.type, ignore_child_nullability))
+        return field.with_nullable(True) if ignore_child_nullability else field
+
     if pa.types.is_large_string(data_type):
         return pa.string()
     if pa.types.is_struct(data_type):
-        return pa.struct([field.with_type(_normalize_string_type(field.type)) for field in data_type])
+        return pa.struct([child(field) for field in data_type])
     if (pa.types.is_list(data_type) or pa.types.is_large_list(data_type)
             or pa.types.is_fixed_size_list(data_type)):
-        field = data_type.value_field.with_type(_normalize_string_type(data_type.value_type))
+        field = child(data_type.value_field)
         if pa.types.is_large_list(data_type):
             return pa.large_list(field)
         if pa.types.is_fixed_size_list(data_type):
@@ -38,13 +42,14 @@ def _normalize_string_type(data_type):
         # includes the sorting flag. Preserve it along with the child fields.
         sorted_keys = data_type == pa.map_(data_type.key_field, data_type.item_field, keys_sorted=True)
         return pa.map_(
-            data_type.key_field.with_type(_normalize_string_type(data_type.key_type)),
-            data_type.item_field.with_type(_normalize_string_type(data_type.item_type)),
+            data_type.key_field.with_type(_normalize_string_type(data_type.key_type, ignore_child_nullability)),
+            child(data_type.item_field),
             keys_sorted=sorted_keys)
     return data_type
 
 
-def arrow_schemas_compatible(source, target, check_top_level_nullability=True, allow_binary_compatibility=False):
+def arrow_schemas_compatible(source, target, check_top_level_nullability=True, allow_binary_compatibility=False,
+                             check_nested_nullability=True):
     """Compare ordered fields, ignoring metadata and STRING offset width.
 
     The writer opts into its existing top-level nullability and binary /
@@ -55,7 +60,9 @@ def arrow_schemas_compatible(source, target, check_top_level_nullability=True, a
     for actual, expected in zip(source, target):
         if check_top_level_nullability and actual.nullable != expected.nullable:
             return False
-        if actual.type == expected.type or _normalize_string_type(actual.type) == _normalize_string_type(expected.type):
+        if (actual.type == expected.type
+                or _normalize_string_type(actual.type, not check_nested_nullability)
+                == _normalize_string_type(expected.type, not check_nested_nullability)):
             continue
         if allow_binary_compatibility and all(
                 pa.types.is_binary(t) or pa.types.is_fixed_size_binary(t)
@@ -65,6 +72,32 @@ def arrow_schemas_compatible(source, target, check_top_level_nullability=True, a
     return True
 
 
+def normalize_arrow_write_nullability(data, target):
+    """Validate read-result NULLs in core before restoring writer child fields."""
+    try:
+        from pypaimon_rust.data import normalize_array_for_write
+    except ImportError:
+        return data
+    options = dict(check_top_level_nullability=False, allow_binary_compatibility=True)
+    if (arrow_schemas_compatible(data.schema, target, **options)
+            or not arrow_schemas_compatible(data.schema, target, check_nested_nullability=False, **options)):
+        return data
+    arrays, fields = [], []
+    for column, field, expected in zip(data.columns, data.schema, target):
+        if column.type != expected.type:
+            if isinstance(column, pa.ChunkedArray):
+                column = pa.chunked_array(
+                    [normalize_array_for_write(chunk, expected.type) for chunk in column.chunks], type=expected.type)
+            else:
+                column = normalize_array_for_write(column, expected.type)
+        arrays.append(column)
+        fields.append(field.with_type(column.type))
+    schema = pa.schema(fields, metadata=data.schema.metadata)
+    if isinstance(data, pa.RecordBatch):
+        return pa.RecordBatch.from_arrays(arrays, schema=schema)
+    return pa.Table.from_arrays(arrays, schema=schema)
+
+
 def normalize_arrow_strings(data):
     """Safely normalize large_string input before routing or buffering rows.