[python] Harden nested schema evolution: reject reshaping casts, fix sliced arrays, gate null-to-not-null

Self-review findings on the nested schema-evolution path:

- update_column_type between same-root constructed types (e.g. ROW<a INT>
  -> ROW<a BIGINT, c STRING>) was accepted: the replacement carried
  caller-supplied nested ids that corrupt the id model and old rows read
  all-NULL; a VECTOR length change was accepted but unreadable. Reject
  non-identical constructed-to-constructed casts - reshaping goes through
  sub-field / 'element' / 'value' paths, which keep working.

- The list/map rebuilds in the alignment and string-rendering paths read
  offsets/raw buffers directly, which errors on a sliced ListArray and
  silently misaligns rows on a sliced MapArray; re-materialize sliced
  inputs first.

- Converting a nullable column to NOT NULL was silently accepted; it is
  now rejected by default and opt-in via
  'alter-column-null-to-not-null.disabled' = 'false'.

Also add an end-to-end test for the array 'element' type promotion path.

Author: TheR1sing3un

paimon-python/pypaimon/casting/data_type_casts.py

@@ -26,8 +26,9 @@
 applies the conversion leniently.
 """
 
-from pypaimon.schema.data_types import (ArrayType, AtomicType, MapType,
-                                        MultisetType, RowType, VectorType)
+from pypaimon.schema.data_types import (ArrayType, AtomicType, DataTypeParser,
+                                        MapType, MultisetType, RowType,
+                                        VectorType)
 
 # ---- Type roots --------------------------------------------------------------
 
@@ -164,9 +165,25 @@ def supports_cast(source_type, target_type, allow_explicit: bool = True) -> bool
     if source_type.nullable and not target_type.nullable and not allow_explicit:
         return False
     if source_root == target_root:
+        if source_root in CONSTRUCTED:
+            # A constructed type is only castable to an (ignoring outer
+            # nullability) identical constructed type. Reshaping is done
+            # through sub-field / 'element' / 'value' paths instead: a whole
+            # ROW replacement would carry caller-supplied nested field ids
+            # that corrupt the id model, and there is no runtime conversion
+            # between differently-shaped constructed values.
+            return _equals_ignore_nullable(source_type, target_type)
         return True
     if source_root in _IMPLICIT_RULES.get(target_root, set()):
         return True
     if allow_explicit and source_root in _EXPLICIT_RULES.get(target_root, set()):
         return True
     return False
+
+
+def _equals_ignore_nullable(source_type, target_type) -> bool:
+    source_copy = DataTypeParser.parse_data_type(source_type.to_dict())
+    target_copy = DataTypeParser.parse_data_type(target_type.to_dict())
+    source_copy.nullable = True
+    target_copy.nullable = True
+    return source_copy == target_copy

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

@@ -37,6 +37,18 @@ def _is_character_string_type(data_type) -> bool:
     return t == 'STRING' or t.startswith('VARCHAR') or t.startswith('CHAR')
 
 
+def _unslice(array):
+    """Re-materialize a sliced array so offsets/buffers start at zero.
+
+    The list/map rebuilds below read ``offsets``/raw buffers directly; on a
+    sliced array those still point into the parent storage, which either
+    errors (list rebuild with a null mask) or silently misaligns rows (map
+    rebuild from raw buffers)."""
+    if array.offset == 0:
+        return array
+    return pa.concat_arrays([array])
+
+
 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)):
@@ -49,6 +61,7 @@ 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."""
+    array = _unslice(array)
     valid = pc.is_valid(array).to_pylist()
     out = []
     if isinstance(file_type, RowType):
@@ -194,11 +207,13 @@ def _align_array_by_id(self, array, file_type, target_type):
             return pa.StructArray.from_arrays(
                 children, fields=pa_fields, mask=pc.is_null(array))
         if isinstance(target_type, ArrayType) and isinstance(file_type, ArrayType):
+            array = _unslice(array)
             aligned_values = self._align_array_by_id(
                 array.values, file_type.element, target_type.element)
             return pa.ListArray.from_arrays(
                 array.offsets, aligned_values, mask=pc.is_null(array))
         if isinstance(target_type, MapType) and isinstance(file_type, MapType):
+            array = _unslice(array)
             aligned_items = self._align_array_by_id(
                 array.items, file_type.value, target_type.value)
             # MapArray.from_arrays cannot carry a null mask (a null map would

paimon-python/pypaimon/schema/schema_manager.py

@@ -181,15 +181,30 @@ def update_func(field: DataField, depth: int) -> DataField:
     _update_nested_column(new_fields, change.field_names, update_func)
 
 
+def _assert_nullability_change(old_nullability: bool, new_nullability: bool,
+                               field_name: str, disable_null_to_not_null: bool):
+    if disable_null_to_not_null and old_nullability and not new_nullability:
+        raise ValueError(
+            "Cannot update column type from nullable to non nullable for {}. "
+            "You can set table configuration option "
+            "'alter-column-null-to-not-null.disabled' = 'false' "
+            "to allow converting null columns to not null".format(field_name)
+        )
+
+
 def _handle_update_column_nullability(
-    change: UpdateColumnNullability, new_fields: List[DataField]
+    change: UpdateColumnNullability, new_fields: List[DataField],
+    disable_null_to_not_null: bool
 ):
     from pypaimon.schema.data_types import DataTypeParser
     field_names = change.field_names
     max_depth = len(field_names)
 
     def update_func(field: DataField, depth: int) -> DataField:
         source_root = _get_root_type(field.type, depth, max_depth)
+        _assert_nullability_change(
+            source_root.nullable, change.new_nullability,
+            '.'.join(field_names), disable_null_to_not_null)
         new_root = DataTypeParser.parse_data_type(source_root.to_dict())
         new_root.nullable = change.new_nullability
         new_type = _get_array_map_type_with_target_type_root(
@@ -652,6 +667,10 @@ def _generate_table_schema(
         # Get add_column_before_partition option
         add_column_before_partition = CoreOptions(Options(old_table_schema.options)).add_column_before_partition()
         partition_keys = old_table_schema.partition_keys
+        # Converting a nullable column to NOT NULL is unsafe for existing
+        # data and is disabled by default; the table option below opts in.
+        disable_null_to_not_null = str(old_table_schema.options.get(
+            'alter-column-null-to-not-null.disabled', 'true')).lower() != 'false'
 
         for change in changes:
             if isinstance(change, SetOption):
@@ -688,7 +707,8 @@ def _generate_table_schema(
                     _assert_not_updating_primary_keys(
                         old_table_schema, change.field_names, "change nullability of"
                     )
-                _handle_update_column_nullability(change, new_fields)
+                _handle_update_column_nullability(
+                    change, new_fields, disable_null_to_not_null)
             elif isinstance(change, UpdateColumnComment):
                 _handle_update_column_comment(change, new_fields)
             elif isinstance(change, UpdateColumnPosition):

paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py

@@ -343,6 +343,31 @@ def test_drop_all_subfields_rejected(self):
                 'default.nsub_dropall',
                 [SchemaChange.drop_column(['mv', 'latest_version'])], False)
 
+    def test_null_to_not_null_disabled_by_default(self):
+        # Converting nullable -> NOT NULL is unsafe for existing data and is
+        # rejected unless the table opts in via
+        # 'alter-column-null-to-not-null.disabled' = 'false'.
+        self._create_struct_table('nsub_nullability')
+        with self.assertRaises(RuntimeError) as cm:
+            self.catalog.alter_table(
+                'default.nsub_nullability',
+                [SchemaChange.update_column_nullability(
+                    ['mv', 'latest_value'], False)], False)
+        self.assertIn('nullable to non nullable', str(cm.exception))
+        # Opt-in makes the same change succeed.
+        self.catalog.alter_table(
+            'default.nsub_nullability',
+            [SchemaChange.set_option(
+                'alter-column-null-to-not-null.disabled', 'false')], False)
+        self.catalog.alter_table(
+            'default.nsub_nullability',
+            [SchemaChange.update_column_nullability(
+                ['mv', 'latest_value'], False)], False)
+        schema = self.catalog.get_table('default.nsub_nullability').table_schema
+        mv = next(f for f in schema.fields if f.name == 'mv')
+        lv = next(sf for sf in mv.type.fields if sf.name == 'latest_value')
+        self.assertFalse(lv.type.nullable)
+
     def test_unsupported_subfield_cast_rejected(self):
         self._create_struct_table('nsub_badcast')
         with self.assertRaises(RuntimeError) as cm:
@@ -476,6 +501,65 @@ def test_array_wrapper_token_validated(self):
             [SchemaChange.add_column(['arr', 'element', 'c'], AtomicType('INT'))],
             False)
 
+    def test_array_element_type_update(self):
+        # The canonical path for promoting an array's element type descends
+        # through the 'element' token; old files are cast at read time.
+        s0 = pa.schema([('id', pa.int64()), ('a2', pa.list_(pa.int32()))])
+        table = self._create('nelem_type', s0)
+        self._write(table, pa.Table.from_pylist(
+            [{'id': 1, 'a2': [1, 2]}], schema=s0))
+        self.catalog.alter_table(
+            'default.nelem_type',
+            [SchemaChange.update_column_type(['a2', 'element'], AtomicType('BIGINT'))],
+            False)
+        table = self.catalog.get_table('default.nelem_type')
+        s1 = pa.schema([('id', pa.int64()), ('a2', pa.list_(pa.int64()))])
+        self._write(table, pa.Table.from_pylist(
+            [{'id': 2, 'a2': [3]}], schema=s1))
+        rb = table.new_read_builder()
+        splits = rb.new_scan().plan().splits()
+        arrow = rb.new_read().to_arrow(splits)
+        self.assertEqual(arrow.schema.field('a2').type, pa.list_(pa.int64()))
+        rows = sorted(arrow.to_pylist(), key=lambda r: r['id'])
+        self.assertEqual(rows, [{'id': 1, 'a2': [1, 2]}, {'id': 2, 'a2': [3]}])
+
+    def test_whole_struct_type_replacement_rejected(self):
+        # Replacing a whole ROW type would carry caller-supplied nested ids
+        # that corrupt the id model; it must be rejected at alter time.
+        elem = pa.struct([('a', pa.int32()), ('b', pa.string())])
+        s0 = pa.schema([('id', pa.int64()), ('mv', elem)])
+        self._create('nrow_replace', s0)
+        new_row = _paimon_type(pa.struct([('a', pa.int64()), ('c', pa.string())]))
+        with self.assertRaises(RuntimeError) as cm:
+            self.catalog.alter_table(
+                'default.nrow_replace',
+                [SchemaChange.update_column_type('mv', new_row)], False)
+        self.assertIn('cannot be converted', str(cm.exception))
+
+    def test_align_handles_sliced_arrays(self):
+        # The list/map rebuilds read offsets/raw buffers; a sliced input
+        # must be re-materialized, not read through stale parent offsets.
+        from pypaimon.read.reader.data_file_batch_reader import \
+            DataFileBatchReader
+        reader = DataFileBatchReader.__new__(DataFileBatchReader)
+        sliced_list = pa.array(
+            [[1, 2], [3], [4, 5, 6], None], type=pa.list_(pa.int32())).slice(1, 3)
+        out = reader._align_array_by_id(
+            sliced_list,
+            ArrayType(True, AtomicType('INT')),
+            ArrayType(True, AtomicType('BIGINT')))
+        self.assertEqual(out.to_pylist(), [[3], [4, 5, 6], None])
+        self.assertEqual(out.type, pa.list_(pa.int64()))
+
+        sliced_map = pa.array(
+            [[('a', 1)], [('b', 2)], None],
+            type=pa.map_(pa.string(), pa.int32())).slice(1, 2)
+        out = reader._align_array_by_id(
+            sliced_map,
+            MapType(True, AtomicType('STRING'), AtomicType('INT')),
+            MapType(True, AtomicType('STRING'), AtomicType('BIGINT')))
+        self.assertEqual(out.to_pylist(), [[('b', 2)], None])
+
     def test_map_wrapper_token_validated(self):
         # The token consumed when descending through a MAP must be 'value'.
         val = pa.struct([('a', pa.int64())])
@@ -657,6 +741,24 @@ def test_constructed_to_string(self):
         for src in (vec, ms):
             self.assertFalse(supports_cast(src, AtomicType('STRING')), str(src))
 
+    def test_constructed_to_differently_shaped_constructed_rejected(self):
+        # Reshaping a constructed type must go through sub-field /
+        # 'element' / 'value' paths; a whole-type replacement would carry
+        # caller-supplied nested ids that corrupt the id model.
+        self.assertFalse(supports_cast(
+            RowType(True, [DataField(0, 'a', AtomicType('INT'))]),
+            RowType(True, [DataField(0, 'a', AtomicType('BIGINT'))])))
+        self.assertFalse(supports_cast(
+            ArrayType(True, AtomicType('INT')),
+            ArrayType(True, AtomicType('BIGINT'))))
+        self.assertFalse(supports_cast(
+            VectorType(True, AtomicType('FLOAT'), 3),
+            VectorType(True, AtomicType('FLOAT'), 5)))
+        # Only the outer nullability differing is still an identity cast.
+        self.assertTrue(supports_cast(
+            RowType(True, [DataField(2, 'a', AtomicType('INT'))]),
+            RowType(False, [DataField(2, 'a', AtomicType('INT'))])))
+
 
 if __name__ == '__main__':
     unittest.main()