Skip to content

Commit 0dccd47

Browse files
TheR1sing3unTheR1sing3un
committed
[python] Support schema evolution of nested struct sub-fields
Read-time schema evolution previously aligned only top-level columns by field id; sub-fields inside a ROW (and a ROW nested in an ARRAY/MAP) could not evolve: adding one silently created a top-level column, and rename/drop/update-type raised because the schema manager only handled the last path element. - Assign globally-unique ids to nested sub-fields at create time and compute highestFieldId recursively, so nested ids never collide with top-level ones. - Recurse schema changes along the dotted field-name path (transparently through ARRAY/MAP wrappers) for add/rename/drop/update-type/update-nullability/ update-comment, allocating new ids from the persisted highestFieldId. - Validate update-column-type against the cast-support rules. - Align nested sub-fields by field id at read time: reorder, pad missing with NULL, follow renames, and cast changed types, recursing into struct/array/map. Add tests covering nested add/rename/drop/update-type round-trips (append-only and primary-key), ARRAY<ROW>/MAP<.,ROW> sub-fields, the id model, and the cast rules.
1 parent 94b468a commit 0dccd47

8 files changed

Lines changed: 750 additions & 162 deletions

File tree

‎paimon-python/pypaimon/casting/__init__.py‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,16 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one
2+
# or more contributor license agreements. See the NOTICE file
3+
# distributed with this work for additional information
4+
# regarding copyright ownership. The ASF licenses this file
5+
# to you under the Apache License, Version 2.0 (the
6+
# "License"); you may not use this file except in compliance
7+
# with the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing,
12+
# software distributed under the License is distributed on an
13+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
# KIND, either express or implied. See the License for the
15+
# specific language governing permissions and limitations
16+
# under the License.

‎paimon-python/pypaimon/casting/data_type_casts.py‎

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one
2+
# or more contributor license agreements. See the NOTICE file
3+
# distributed with this work for additional information
4+
# regarding copyright ownership. The ASF licenses this file
5+
# to you under the Apache License, Version 2.0 (the
6+
# "License"); you may not use this file except in compliance
7+
# with the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing,
12+
# software distributed under the License is distributed on an
13+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
# KIND, either express or implied. See the License for the
15+
# specific language governing permissions and limitations
16+
# under the License.
17+
18+
"""Type-cast support rules used to validate ``update column type`` schema
19+
changes.
20+
21+
The rules mirror the engine-wide cast specification so a type change accepted
22+
here is one the read path can also materialize: an *implicit* cast is a safe
23+
widening (e.g. INT -> BIGINT, any numeric -> DECIMAL/DOUBLE), while an
24+
*explicit* cast covers the broader, possibly lossy conversions a user opts into
25+
(e.g. DOUBLE -> INT truncation, anything -> STRING). Read-time execution then
26+
applies the conversion leniently.
27+
"""
28+
29+
from pypaimon.schema.data_types import (ArrayType, AtomicType, MapType,
30+
MultisetType, RowType, VectorType)
31+
32+
# ---- Type roots --------------------------------------------------------------
33+
34+
CHAR = "CHAR"
35+
VARCHAR = "VARCHAR"
36+
BOOLEAN = "BOOLEAN"
37+
BINARY = "BINARY"
38+
VARBINARY = "VARBINARY"
39+
DECIMAL = "DECIMAL"
40+
TINYINT = "TINYINT"
41+
SMALLINT = "SMALLINT"
42+
INTEGER = "INTEGER"
43+
BIGINT = "BIGINT"
44+
FLOAT = "FLOAT"
45+
DOUBLE = "DOUBLE"
46+
DATE = "DATE"
47+
TIME = "TIME"
48+
TIMESTAMP = "TIMESTAMP"
49+
TIMESTAMP_LTZ = "TIMESTAMP_LTZ"
50+
ARRAY = "ARRAY"
51+
MAP = "MAP"
52+
MULTISET = "MULTISET"
53+
ROW = "ROW"
54+
VECTOR = "VECTOR"
55+
VARIANT = "VARIANT"
56+
BLOB = "BLOB"
57+
58+
# ---- Families ----------------------------------------------------------------
59+
60+
CHARACTER_STRING = {CHAR, VARCHAR}
61+
BINARY_STRING = {BINARY, VARBINARY}
62+
INTEGER_NUMERIC = {TINYINT, SMALLINT, INTEGER, BIGINT}
63+
NUMERIC = INTEGER_NUMERIC | {FLOAT, DOUBLE, DECIMAL}
64+
TIMESTAMP_FAMILY = {TIMESTAMP, TIMESTAMP_LTZ}
65+
TIME_FAMILY = {TIME}
66+
DATETIME = {DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ}
67+
PREDEFINED = {
68+
CHAR, VARCHAR, BOOLEAN, BINARY, VARBINARY, DECIMAL,
69+
TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE,
70+
DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ,
71+
}
72+
CONSTRUCTED = {ARRAY, MAP, MULTISET, ROW, VECTOR}
73+
74+
75+
def _root(data_type) -> str:
76+
if isinstance(data_type, RowType):
77+
return ROW
78+
if isinstance(data_type, ArrayType):
79+
return ARRAY
80+
if isinstance(data_type, MapType):
81+
return MAP
82+
if isinstance(data_type, MultisetType):
83+
return MULTISET
84+
if isinstance(data_type, VectorType):
85+
return VECTOR
86+
if isinstance(data_type, AtomicType):
87+
t = data_type.type.upper()
88+
if t.startswith("DECIMAL") or t.startswith("NUMERIC") or t.startswith("DEC"):
89+
return DECIMAL
90+
if t in ("INT", "INTEGER"):
91+
return INTEGER
92+
if t in (TINYINT, SMALLINT, BIGINT, FLOAT, DOUBLE, BOOLEAN, DATE):
93+
return t
94+
if t == "STRING" or t.startswith("VARCHAR"):
95+
return VARCHAR
96+
if t.startswith("CHAR"):
97+
return CHAR
98+
if t == "BYTES" or t.startswith("VARBINARY"):
99+
return VARBINARY
100+
if t.startswith("BINARY"):
101+
return BINARY
102+
if t == "BLOB":
103+
return BLOB
104+
if t.startswith("TIMESTAMP_LTZ"):
105+
return TIMESTAMP_LTZ
106+
if t.startswith("TIMESTAMP"):
107+
return TIMESTAMP
108+
if t.startswith("TIME"):
109+
return TIME
110+
if t == "VARIANT":
111+
return VARIANT
112+
return None
113+
114+
115+
def _build_rules():
116+
implicit = {}
117+
explicit = {}
118+
# Identity cast for every root.
119+
for root in (PREDEFINED | CONSTRUCTED | {VARIANT, BLOB}):
120+
implicit[root] = {root}
121+
explicit[root] = set()
122+
123+
def rule(target, implicit_from=None, explicit_from=None):
124+
implicit[target] |= set(implicit_from or set())
125+
explicit[target] |= set(explicit_from or set())
126+
127+
rule(CHAR, {CHAR}, PREDEFINED | CONSTRUCTED)
128+
rule(VARCHAR, CHARACTER_STRING, PREDEFINED | CONSTRUCTED)
129+
rule(BOOLEAN, {BOOLEAN}, CHARACTER_STRING | INTEGER_NUMERIC)
130+
rule(BINARY, {BINARY}, CHARACTER_STRING | {VARBINARY})
131+
rule(VARBINARY, BINARY_STRING, CHARACTER_STRING | {BINARY})
132+
rule(DECIMAL, NUMERIC, CHARACTER_STRING | {BOOLEAN, TIMESTAMP, TIMESTAMP_LTZ})
133+
int_explicit = NUMERIC | CHARACTER_STRING | {BOOLEAN, TIMESTAMP, TIMESTAMP_LTZ}
134+
rule(TINYINT, {TINYINT}, int_explicit)
135+
rule(SMALLINT, {TINYINT, SMALLINT}, int_explicit)
136+
rule(INTEGER, {TINYINT, SMALLINT, INTEGER}, int_explicit)
137+
rule(BIGINT, {TINYINT, SMALLINT, INTEGER, BIGINT}, int_explicit)
138+
rule(FLOAT, {TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DECIMAL}, int_explicit)
139+
rule(DOUBLE, NUMERIC, CHARACTER_STRING | {BOOLEAN, TIMESTAMP, TIMESTAMP_LTZ})
140+
rule(DATE, {DATE, TIMESTAMP}, TIMESTAMP_FAMILY | CHARACTER_STRING)
141+
rule(TIME, {TIME, TIMESTAMP}, TIME_FAMILY | TIMESTAMP_FAMILY | CHARACTER_STRING)
142+
rule(TIMESTAMP, {TIMESTAMP, TIMESTAMP_LTZ}, DATETIME | CHARACTER_STRING | NUMERIC)
143+
rule(TIMESTAMP_LTZ, {TIMESTAMP_LTZ, TIMESTAMP}, DATETIME | CHARACTER_STRING | NUMERIC)
144+
return implicit, explicit
145+
146+
147+
_IMPLICIT_RULES, _EXPLICIT_RULES = _build_rules()
148+
149+
150+
def supports_cast(source_type, target_type, allow_explicit: bool = True) -> bool:
151+
"""Whether ``source_type`` can be cast to ``target_type`` for a column type
152+
change. ``allow_explicit`` permits the broader (possibly lossy) conversions
153+
in addition to the safe widening ones."""
154+
source_root = _root(source_type)
155+
target_root = _root(target_type)
156+
if source_root is None or target_root is None:
157+
return False
158+
# A NOT NULL target cannot accept a nullable source unless explicitly allowed.
159+
if source_type.nullable and not target_type.nullable and not allow_explicit:
160+
return False
161+
if source_root == target_root:
162+
return True
163+
if source_root in _IMPLICIT_RULES.get(target_root, set()):
164+
return True
165+
if allow_explicit and source_root in _EXPLICIT_RULES.get(target_root, set()):
166+
return True
167+
return False

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

Lines changed: 79 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -18,13 +18,15 @@
1818
from typing import List, Optional
1919

2020
import pyarrow as pa
21+
import pyarrow.compute as pc
2122
from pyarrow import RecordBatch
2223

2324
from pypaimon.common.file_io import FileIO
2425
from pypaimon.read.partition_info import PartitionInfo
2526
from pypaimon.read.reader.format_blob_reader import FormatBlobReader
2627
from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
27-
from pypaimon.schema.data_types import DataField, PyarrowFieldParser
28+
from pypaimon.schema.data_types import (ArrayType, DataField, MapType,
29+
PyarrowFieldParser, RowType)
2830
from pypaimon.table.special_fields import SpecialFields
2931

3032

@@ -57,55 +59,99 @@ def __init__(self, format_reader: RecordBatchReader, index_mapping: List[int], p
5759
self.file_io = file_io
5860
# Per-file field-id normalization: map the physically-read columns
5961
# (the file's own field order/names) onto the latest read target by
60-
# field id, padding missing ids with NULL. ``None`` when there is no
61-
# evolution to reconcile (identity) -- the common path stays zero-copy.
62-
self._normalize_positions, self._normalize_names = \
63-
self._build_normalize_plan(file_data_fields, target_data_fields)
62+
# field id, padding missing ids with NULL and recursing into nested
63+
# ROW / ARRAY<ROW> / MAP<.,ROW> sub-fields the same way. ``None`` when
64+
# there is no evolution to reconcile -- the common path stays zero-copy.
65+
self._normalize_plan = self._build_normalize_plan(file_data_fields, target_data_fields)
6466

6567
@staticmethod
6668
def _build_normalize_plan(file_data_fields, target_data_fields):
6769
"""Build a per-file field-id alignment plan.
6870
69-
Returns ``(positions, names)`` where ``positions[i]`` is the column
70-
index in the physically-read batch carrying ``target_data_fields[i]``
71-
(matched by field id), or -1 if the file does not contain that id (pad
72-
NULL). ``names[i]`` is the latest target name. Returns ``(None, None)``
73-
when the plan is the identity (no evolution), so the caller skips
74-
normalization and stays zero-copy.
71+
Returns a list of ``(pos, file_field, target_field)`` -- one per target
72+
field, in target order -- where ``pos`` is the column index in the
73+
physically-read batch carrying ``target_field`` (matched by field id),
74+
or -1 if the file does not contain that id (pad NULL). Returns ``None``
75+
when the file already matches the target exactly (no evolution), so the
76+
caller stays zero-copy.
7577
"""
7678
if file_data_fields is None or target_data_fields is None:
77-
return None, None
79+
return None
80+
# Recursive equality covers nested sub-field changes too: any rename /
81+
# add / drop / type change at any depth makes the file != target.
82+
if file_data_fields == target_data_fields:
83+
return None
7884
file_id_to_pos = {f.id: i for i, f in enumerate(file_data_fields)}
79-
positions = []
80-
names = []
81-
# Identity only when every target maps to the same physical position
82-
# AND already carries the same name -- a rename keeps the position but
83-
# changes the name, which still requires a relabel pass.
84-
identity = len(file_data_fields) == len(target_data_fields)
85-
for i, target in enumerate(target_data_fields):
85+
plan = []
86+
for target in target_data_fields:
8687
pos = file_id_to_pos.get(target.id, -1)
87-
positions.append(pos)
88-
names.append(target.name)
89-
if pos != i or (pos >= 0 and file_data_fields[pos].name != target.name):
90-
identity = False
91-
if identity:
92-
return None, None
93-
return positions, names
88+
file_field = file_data_fields[pos] if pos >= 0 else None
89+
plan.append((pos, file_field, target))
90+
return plan
9491

9592
def _normalize_batch(self, record_batch: RecordBatch) -> RecordBatch:
9693
"""Reorder/pad the physically-read batch onto the latest read target by
97-
field id, and relabel columns to the latest names. Missing ids become
98-
all-NULL columns; types are reconciled later by _align_batch_to_read_schema."""
99-
if self._normalize_positions is None:
94+
field id, relabel columns to the latest names, and align nested ROW
95+
sub-fields by id. Missing ids become typed all-NULL columns."""
96+
if self._normalize_plan is None:
10097
return record_batch
10198
num_rows = record_batch.num_rows
10299
arrays = []
103-
for pos in self._normalize_positions:
100+
names = []
101+
for pos, file_field, target_field in self._normalize_plan:
102+
target_pa_type = PyarrowFieldParser.from_paimon_type(target_field.type)
104103
if pos < 0:
105-
arrays.append(pa.nulls(num_rows))
104+
arrays.append(pa.nulls(num_rows, type=target_pa_type))
106105
else:
107-
arrays.append(record_batch.column(pos))
108-
return pa.RecordBatch.from_arrays(arrays, names=self._normalize_names)
106+
arrays.append(self._align_array_by_id(
107+
record_batch.column(pos), file_field.type, target_field.type))
108+
names.append(target_field.name)
109+
return pa.RecordBatch.from_arrays(arrays, names=names)
110+
111+
def _align_array_by_id(self, array, file_type, target_type):
112+
"""Return *array* converted to *target_type*, matching ROW sub-fields by
113+
field id (reorder, pad missing with NULL, follow renames, cast changed
114+
types) recursively, transparently through ARRAY/MAP wrappers."""
115+
if isinstance(target_type, RowType) and isinstance(file_type, RowType):
116+
n = len(array)
117+
file_id_to_pos = {f.id: i for i, f in enumerate(file_type.fields)}
118+
children = []
119+
pa_fields = []
120+
for tsub in target_type.fields:
121+
p = file_id_to_pos.get(tsub.id, -1)
122+
if p < 0:
123+
child = pa.nulls(n, type=PyarrowFieldParser.from_paimon_type(tsub.type))
124+
else:
125+
child = self._align_array_by_id(
126+
array.field(p), file_type.fields[p].type, tsub.type)
127+
children.append(child)
128+
pa_fields.append(pa.field(tsub.name, child.type, nullable=tsub.type.nullable))
129+
# Preserve the struct's own null mask; child values under a null
130+
# struct are irrelevant.
131+
return pa.StructArray.from_arrays(
132+
children, fields=pa_fields, mask=pc.is_null(array))
133+
if isinstance(target_type, ArrayType) and isinstance(file_type, ArrayType):
134+
aligned_values = self._align_array_by_id(
135+
array.values, file_type.element, target_type.element)
136+
return pa.ListArray.from_arrays(
137+
array.offsets, aligned_values, mask=pc.is_null(array))
138+
if isinstance(target_type, MapType) and isinstance(file_type, MapType):
139+
aligned_items = self._align_array_by_id(
140+
array.items, file_type.value, target_type.value)
141+
# MapArray.from_arrays cannot carry a null mask (a null map would
142+
# collapse to an empty one), so rebuild from buffers, reusing the
143+
# original validity/offset buffers and only swapping the value child.
144+
target_pa = PyarrowFieldParser.from_paimon_type(target_type)
145+
entries = pa.StructArray.from_arrays(
146+
[array.keys, aligned_items],
147+
fields=[target_pa.key_field, target_pa.item_field])
148+
return pa.Array.from_buffers(
149+
target_pa, len(array), array.buffers()[:2], children=[entries])
150+
# Leaf / non-nested: cast to the target type when it differs.
151+
target_pa_type = PyarrowFieldParser.from_paimon_type(target_type)
152+
if array.type != target_pa_type:
153+
return array.cast(target_pa_type, safe=False)
154+
return array
109155

110156
def read_arrow_batch(self, start_idx=None, end_idx=None) -> Optional[RecordBatch]:
111157
if isinstance(self.format_reader, FormatBlobReader):

0 commit comments

Comments
 (0)