[python] Reject column type casts the read path cannot execute

supports_cast only encodes the logical cast specification (mirroring Java
DataTypeCasts), so casts with no runtime implementation -- e.g. TIMESTAMP ->
DECIMAL, BOOLEAN -> DECIMAL, TIME -> TIMESTAMP -- were accepted at alter time
and then failed at read with ArrowNotImplementedError. Java additionally checks
CastExecutors.resolve(...) != null.

Add can_execute_cast as the executable-cast counterpart: leaf casts defer to
PyArrow's cast-kernel availability (probed once and cached), constructed ->
string is rendered by the read path, and other constructed conversions are
rejected. update column type now requires both supports_cast and
can_execute_cast.

Author: TheR1sing3un

paimon-python/pypaimon/casting/data_type_casts.py

@@ -26,8 +26,11 @@
 applies the conversion leniently.
 """
 
+import pyarrow as pa
+
 from pypaimon.schema.data_types import (ArrayType, AtomicType, DataTypeParser,
-                                        MapType, MultisetType, RowType,
+                                        MapType, MultisetType,
+                                        PyarrowFieldParser, RowType,
                                         VectorType)
 
 # ---- Type roots --------------------------------------------------------------
@@ -187,3 +190,63 @@ def _equals_ignore_nullable(source_type, target_type) -> bool:
     source_copy.nullable = True
     target_copy.nullable = True
     return source_copy == target_copy
+
+
+# Caches the PyArrow cast-kernel probe per (source, target) pyarrow type so the
+# alter-time check stays cheap. Keyed by the pyarrow type strings.
+_EXECUTABLE_CAST_CACHE = {}
+
+
+def can_execute_cast(source_type, target_type) -> bool:
+    """Whether the Python read path can actually *materialize* a stored
+    ``source_type`` value as ``target_type`` when reading a file written before
+    the column type change.
+
+    ``supports_cast`` only encodes the *logical* cast specification (mirroring
+    Java ``DataTypeCasts``). This is the executable-cast counterpart of Java's
+    ``CastExecutors.resolve(...) != null`` guard: some logically-valid casts
+    (e.g. ``TIMESTAMP -> DECIMAL``, ``BOOLEAN -> DECIMAL``, ``TIME ->
+    TIMESTAMP``) have no PyArrow cast kernel, so without this check the alter
+    succeeds and the read later fails with ``ArrowNotImplementedError``.
+    """
+    source_root = _root(source_type)
+    target_root = _root(target_type)
+    if source_root is None or target_root is None:
+        return False
+    # 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:
+        return True
+    # Constructed -> character string is rendered by the read path's custom
+    # ``_constructed_to_string_array`` (see DataFileBatchReader), not a cast.
+    if (source_root in STRING_RENDERABLE_CONSTRUCTED
+            and target_root in CHARACTER_STRING):
+        return True
+    # Any other conversion touching a constructed type has no runtime cast.
+    if source_root in CONSTRUCTED or target_root in CONSTRUCTED:
+        return False
+    # Leaf-to-leaf: defer to PyArrow's cast-kernel availability, which is the
+    # read path's actual cast executor (``array.cast(target, safe=False)``).
+    return _pyarrow_cast_supported(source_type, target_type)
+
+
+def _pyarrow_cast_supported(source_type, target_type) -> bool:
+    source_pa = PyarrowFieldParser.from_paimon_type(source_type)
+    target_pa = PyarrowFieldParser.from_paimon_type(target_type)
+    if source_pa == target_pa:
+        return True
+    cache_key = (str(source_pa), str(target_pa))
+    cached = _EXECUTABLE_CAST_CACHE.get(cache_key)
+    if cached is not None:
+        return cached
+    # An empty-array cast exercises only kernel resolution (not per-value
+    # conversion), so it reports whether PyArrow has a cast kernel for the pair
+    # without depending on any data. ``safe=False`` matches the read path.
+    try:
+        pa.array([], type=source_pa).cast(target_pa, safe=False)
+        ok = True
+    except (pa.lib.ArrowNotImplementedError, pa.lib.ArrowInvalid,
+            pa.lib.ArrowTypeError):
+        ok = False
+    _EXECUTABLE_CAST_CACHE[cache_key] = ok
+    return ok

paimon-python/pypaimon/schema/schema_manager.py

@@ -26,7 +26,7 @@
 from pypaimon.schema.column_directive_utils import (
     apply_add_column_directive, apply_directives,
     remove_dropped_directive_options)
-from pypaimon.casting.data_type_casts import supports_cast
+from pypaimon.casting.data_type_casts import can_execute_cast, supports_cast
 from pypaimon.schema.data_types import (ArrayType, AtomicInteger, DataField,
                                         MapType, RowType, reassign_field_id)
 from pypaimon.schema.schema import Schema
@@ -240,6 +240,15 @@ def update_func(field: DataField, depth: int) -> DataField:
                 "Column type {}[{}] cannot be converted to {} without losing information."
                 .format(field.name, source_root, target_root)
             )
+        # Logical cast support is not enough: the read path materializes the
+        # change via PyArrow when reading old files, so reject casts it cannot
+        # execute (mirrors Java's CastExecutors.resolve(...) != null check).
+        if not can_execute_cast(source_root, target_root):
+            raise ValueError(
+                "Column type {}[{}] cannot be converted to {}: the read path "
+                "has no executable cast for this conversion."
+                .format(field.name, source_root, target_root)
+            )
         new_type = _get_array_map_type_with_target_type_root(
             field.type, target_root, depth, max_depth)
         return DataField(

paimon-python/pypaimon/tests/filesystem_catalog_test.py

@@ -257,6 +257,40 @@ def _make_table(name, options):
         table = catalog.get_table(allowed_id)
         self.assertFalse(table.fields[1].type.nullable)
 
+    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"
+        schema = Schema(
+            fields=[
+                DataField.from_dict({"id": 0, "name": "k", "type": "INT"}),
+                DataField.from_dict({"id": 1, "name": "ts", "type": "TIMESTAMP(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))
+
+        # An executable widening still succeeds.
+        catalog.alter_table(
+            identifier,
+            [SchemaChange.update_column_type("k", AtomicType("BIGINT"))],
+            False)
+        table = catalog.get_table(identifier)
+        self.assertEqual(table.fields[0].type.type, "BIGINT")
+
     def test_add_column_before_partition(self):
         catalog = CatalogFactory.create({
             "warehouse": self.warehouse