[python] Materialize constructed-type to STRING casts at read time

TheR1sing3un · TheR1sing3un · commit 824faae35a95 · 2026-06-11T10:52:26.000+08:00
update_column_type from ROW/ARRAY/MAP to STRING passes validation (the
cast rules allow constructed types to character strings), but reading an
old file failed with ArrowNotImplementedError because struct/list/map
cannot be cast to utf8 directly.

Render the string form during per-file alignment instead, matching the
engine's cast rules: ROW as '{v1, v2}', ARRAY as '[e1, e2]', MAP as
'{k1 -&gt; v1, k2 -&gt; v2}', with sub-values rendered recursively, NULL
sub-values as the literal 'null', and NULL containers staying NULL.

Add round-trip tests for ROW/ARRAY/MAP to STRING, NULL semantics, and a
nested sub-field changed to STRING.
diff --git a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
@@ -25,11 +25,74 @@
 from pypaimon.read.partition_info import PartitionInfo
 from pypaimon.read.reader.format_blob_reader import FormatBlobReader
 from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
-from pypaimon.schema.data_types import (ArrayType, DataField, MapType,
-                                        PyarrowFieldParser, RowType)
+from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField,
+                                        MapType, PyarrowFieldParser, RowType)
 from pypaimon.table.special_fields import SpecialFields
 
 
+def _is_character_string_type(data_type) -> bool:
+    if not isinstance(data_type, AtomicType):
+        return False
+    t = data_type.type.upper()
+    return t == 'STRING' or t.startswith('VARCHAR') or t.startswith('CHAR')
+
+
+def _to_string_values(array, data_type) -> list:
+    """Render *array* as a list of per-row strings (None for NULL rows)."""
+    if isinstance(data_type, (RowType, ArrayType, MapType)):
+        return _constructed_to_string_array(array, data_type).to_pylist()
+    return array.cast(pa.string(), safe=False).to_pylist()
+
+
+def _constructed_to_string_array(array, file_type):
+    """Render a struct/list/map array in the engine's string form:
+    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."""
+    valid = pc.is_valid(array).to_pylist()
+    out = []
+    if isinstance(file_type, RowType):
+        children = [
+            _to_string_values(array.field(i), sub.type)
+            for i, sub in enumerate(file_type.fields)
+        ]
+        for i in range(len(array)):
+            if not valid[i]:
+                out.append(None)
+                continue
+            vals = [c[i] if c[i] is not None else 'null' for c in children]
+            out.append('{' + ', '.join(vals) + '}')
+    elif isinstance(file_type, ArrayType):
+        values = _to_string_values(array.values, file_type.element)
+        offsets = array.offsets.to_pylist()
+        for i in range(len(array)):
+            if not valid[i]:
+                out.append(None)
+                continue
+            elems = [v if v is not None else 'null'
+                     for v in values[offsets[i]:offsets[i + 1]]]
+            out.append('[' + ', '.join(elems) + ']')
+    elif isinstance(file_type, MapType):
+        keys = _to_string_values(array.keys, file_type.key)
+        items = _to_string_values(array.items, file_type.value)
+        offsets = array.offsets.to_pylist()
+        for i in range(len(array)):
+            if not valid[i]:
+                out.append(None)
+                continue
+            entries = [
+                '{} -> {}'.format(
+                    keys[j] if keys[j] is not None else 'null',
+                    items[j] if items[j] is not None else 'null')
+                for j in range(offsets[i], offsets[i + 1])
+            ]
+            out.append('{' + ', '.join(entries) + '}')
+    else:
+        raise ValueError(
+            'Unsupported constructed type for string rendering: {}'.format(file_type))
+    return pa.array(out, type=pa.string())
+
+
 class DataFileBatchReader(RecordBatchReader):
     """
     Reads record batch from files of different formats
@@ -147,6 +210,12 @@ def _align_array_by_id(self, array, file_type, target_type):
                 fields=[target_pa.key_field, target_pa.item_field])
             return pa.Array.from_buffers(
                 target_pa, len(array), array.buffers()[:2], children=[entries])
+        # A constructed type changed to a character string: pyarrow cannot
+        # cast struct/list/map to utf8 directly, so render the engine's
+        # string form instead.
+        if (isinstance(file_type, (RowType, ArrayType, MapType))
+                and _is_character_string_type(target_type)):
+            return _constructed_to_string_array(array, file_type)
         # Leaf / non-nested: cast to the target type when it differs.
         target_pa_type = PyarrowFieldParser.from_paimon_type(target_type)
         if array.type != target_pa_type:
diff --git a/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py b/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
@@ -458,6 +458,88 @@ def test_map_of_row_add_subfield(self):
         self.assertEqual(rows[0]['m'], [('k', {'a': 1, 'b': 'x', 'c': None})])
 
 
+class SchemaEvolutionConstructedToStringTest(_NestedBase):
+    """update column type from ROW/ARRAY/MAP to STRING: old files must be
+    materialized as the engine's string rendering at read time."""
+
+    def test_row_to_string(self):
+        s0 = pa.schema([('id', pa.int64()),
+                        ('mv', pa.struct([('a', pa.int32()), ('b', pa.string())]))])
+        table = self._create('c2s_row', s0)
+        self._write(table, pa.Table.from_pylist(
+            [{'id': 1, 'mv': {'a': 1, 'b': 'x'}}], schema=s0))
+        self.catalog.alter_table(
+            'default.c2s_row',
+            [SchemaChange.update_column_type('mv', AtomicType('STRING'))], False)
+        table = self.catalog.get_table('default.c2s_row')
+        s1 = pa.schema([('id', pa.int64()), ('mv', pa.string())])
+        self._write(table, pa.Table.from_pylist(
+            [{'id': 2, 'mv': 's2'}], schema=s1))
+
+        rows = self._read_sorted(table)
+        self.assertEqual(rows, [
+            {'id': 1, 'mv': '{1, x}'},
+            {'id': 2, 'mv': 's2'},
+        ])
+
+    def test_array_to_string(self):
+        s0 = pa.schema([('id', pa.int64()), ('arr', pa.list_(pa.int32()))])
+        table = self._create('c2s_arr', s0)
+        self._write(table, pa.Table.from_pylist(
+            [{'id': 1, 'arr': [1, 2, 3]}], schema=s0))
+        self.catalog.alter_table(
+            'default.c2s_arr',
+            [SchemaChange.update_column_type('arr', AtomicType('STRING'))], False)
+        table = self.catalog.get_table('default.c2s_arr')
+        rows = self._read_sorted(table)
+        self.assertEqual(rows[0]['arr'], '[1, 2, 3]')
+
+    def test_map_to_string(self):
+        s0 = pa.schema([('id', pa.int64()),
+                        ('m', pa.map_(pa.string(), pa.int32()))])
+        table = self._create('c2s_map', s0)
+        self._write(table, pa.Table.from_pylist(
+            [{'id': 1, 'm': [('k', 7)]}], schema=s0))
+        self.catalog.alter_table(
+            'default.c2s_map',
+            [SchemaChange.update_column_type('m', AtomicType('STRING'))], False)
+        table = self.catalog.get_table('default.c2s_map')
+        rows = self._read_sorted(table)
+        self.assertEqual(rows[0]['m'], '{k -> 7}')
+
+    def test_row_to_string_null_semantics(self):
+        s0 = pa.schema([('id', pa.int64()),
+                        ('mv', pa.struct([('a', pa.int32()), ('b', pa.string())]))])
+        table = self._create('c2s_null', s0)
+        self._write(table, pa.Table.from_pylist([
+            {'id': 1, 'mv': None},
+            {'id': 2, 'mv': {'a': None, 'b': 'x'}},
+        ], schema=s0))
+        self.catalog.alter_table(
+            'default.c2s_null',
+            [SchemaChange.update_column_type('mv', AtomicType('STRING'))], False)
+        table = self.catalog.get_table('default.c2s_null')
+        rows = self._read_sorted(table)
+        # A NULL container stays NULL; a NULL sub-value renders as 'null'.
+        self.assertIsNone(rows[0]['mv'])
+        self.assertEqual(rows[1]['mv'], '{null, x}')
+
+    def test_nested_subfield_row_to_string(self):
+        inner = pa.struct([('a', pa.int32())])
+        s0 = pa.schema([('id', pa.int64()),
+                        ('mv', pa.struct([('inner', inner)]))])
+        table = self._create('c2s_nested', s0)
+        self._write(table, pa.Table.from_pylist(
+            [{'id': 1, 'mv': {'inner': {'a': 1}}}], schema=s0))
+        self.catalog.alter_table(
+            'default.c2s_nested',
+            [SchemaChange.update_column_type(['mv', 'inner'], AtomicType('STRING'))],
+            False)
+        table = self.catalog.get_table('default.c2s_nested')
+        rows = self._read_sorted(table)
+        self.assertEqual(rows[0]['mv'], {'inner': '{1}'})
+
+
 class NestedFieldIdModelTest(unittest.TestCase):
     """Globally-unique nested field ids, mirrored from the engine id model."""
 
