This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 4941e5d528 [python] Delegate schema evolution and runtime read schemas
to Rust core (#10572)
4941e5d528 is described below
commit 4941e5d5280f1e38ed5e1a3ca2c48a2a2b798b91
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Oct 15 10:24:45 2026 +0800
[python] Delegate schema evolution and runtime read schemas to Rust core
(#10572)
---
.github/workflows/ci-python.yml | 9 +-
paimon-python/pypaimon/casting/data_type_casts.py | 7 +-
paimon-python/pypaimon/casting/schema_evolution.py | 44 +++-
.../pypaimon/read/reader/data_file_batch_reader.py | 13 +-
paimon-python/pypaimon/read/table_read.py | 24 +-
paimon-python/pypaimon/schema/arrow_schema.py | 47 +++-
.../tests/native_constructed_evolution_test.py | 96 ++++++++
.../pypaimon/tests/native_read_type_test.py | 21 ++
.../pypaimon/tests/native_schema_cast_test.py | 243 +++++++++++++++++++++
.../tests/py36/filter_record_batch_reader_test.py | 64 ++++++
.../pypaimon/tests/schema_evolution_cast_test.py | 7 +
.../tests/schema_evolution_nested_read_test.py | 12 +-
paimon-python/pypaimon/write/native_write.py | 8 +-
paimon-python/pypaimon/write/table_write.py | 20 +-
14 files changed, 584 insertions(+), 31 deletions(-)
diff --git a/.github/workflows/ci-python.yml b/.github/workflows/ci-python.yml
index c28becd9f5..797bed8bf6 100644
--- a/.github/workflows/ci-python.yml
+++ b/.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:
diff --git a/paimon-python/pypaimon/casting/data_type_casts.py
b/paimon-python/pypaimon/casting/data_type_casts.py
index 849be57e54..027f3293c7 100644
--- a/paimon-python/pypaimon/casting/data_type_casts.py
+++ b/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
diff --git a/paimon-python/pypaimon/casting/schema_evolution.py
b/paimon-python/pypaimon/casting/schema_evolution.py
index 1116aeaacd..1b6c76175d 100644
--- a/paimon-python/pypaimon/casting/schema_evolution.py
+++ b/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:
diff --git a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
index e9d8066a39..3e5c2d6e3a 100644
--- a/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
+++ b/paimon-python/pypaimon/read/reader/data_file_batch_reader.py
@@ -21,7 +21,7 @@ import pyarrow as pa
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 @@ class DataFileBatchReader(RecordBatchReader):
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 @@ class DataFileBatchReader(RecordBatchReader):
"""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)}
diff --git a/paimon-python/pypaimon/read/table_read.py
b/paimon-python/pypaimon/read/table_read.py
index 537c2ae7b2..6e0b26362f 100644
--- a/paimon-python/pypaimon/read/table_read.py
+++ b/paimon-python/pypaimon/read/table_read.py
@@ -25,6 +25,8 @@ from typing import Any, Callable, Dict, Iterator, List,
Optional
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 @@ from pypaimon.read.split_read import (DataEvolutionSplitRead,
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 @@ class TableRead:
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 @@ class TableRead:
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 @@ class TableRead:
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 @@ class TableRead:
# 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 @@ class TableRead:
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 (
diff --git a/paimon-python/pypaimon/schema/arrow_schema.py
b/paimon-python/pypaimon/schema/arrow_schema.py
index 98a232033b..010b1ae2d8 100644
--- a/paimon-python/pypaimon/schema/arrow_schema.py
+++ b/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.
diff --git a/paimon-python/pypaimon/tests/native_constructed_evolution_test.py
b/paimon-python/pypaimon/tests/native_constructed_evolution_test.py
new file mode 100644
index 0000000000..2eab7af726
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_constructed_evolution_test.py
@@ -0,0 +1,96 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Existing constructed schema changes must read physical Arrow lists
natively."""
+
+import os
+from contextlib import ExitStack
+from unittest.mock import patch
+from urllib.parse import unquote, urlparse
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.schema_change import SchemaChange
+from pypaimon.tests import native_schema_evolution_cast_test
+
+rest_catalog = native_schema_evolution_cast_test.rest_catalog
+pytestmark = native_schema_evolution_cast_test.pytestmark
+
+_LIST = pa.list_(pa.int32())
+_LARGE_LIST = pa.large_list(pa.int32())
+_CASES = [
+ (_LIST, _LARGE_LIST, [[9], [1, None], [], None, [2, 3]], 'v',
+ ['[9]', '[1, null]', '[]', None, '[2, 3]']),
+ (pa.struct([('a', _LIST)]), pa.struct([('a', _LARGE_LIST)]),
+ [{'a': None}, None, {'a': [1, None]}, {'a': []}], 'v',
+ ['{null}', None, '{[1, null]}', '{[]}']),
+ (pa.map_(pa.string(), _LIST), pa.map_(pa.string(), _LARGE_LIST),
+ [[('one', [1, None]), ('empty', []), ('missing', None)], None, []], 'v',
+ ['{one -> [1, null], empty -> [], missing -> null}', None, '{}']),
+ (pa.struct([('a', _LIST)]), pa.struct([('a', _LARGE_LIST)]),
+ [{'a': None}, None, {'a': [1, None]}, {'a': []}], ['v', 'a'],
+ [{'a': None}, None, {'a': '[1, null]'}, {'a': '[]'}]),
+]
+
+
[email protected]('mode', ['append', 'pk', 'de'])
[email protected]('target', ['STRING', 'CHAR(2)'])
[email protected]('source,physical,values,path,expected', _CASES,
+ ids=['array', 'row', 'map', 'row-child'])
+def test_large_list_constructed_strings_use_native_read(
+ rest_catalog, mode, target, source, physical, values, path, expected):
+ catalog, _ = rest_catalog
+ native_schema_evolution_cast_test._table(catalog, mode, source, values)
+ # The original nested schema must also plan through Rust, before changing
+ # only the logical type and preserving the old physical data files.
+ original =
catalog.get_table('default.t').copy({'scan.native-plan.enabled': 'true'})
+ original_plan = original.new_read_builder().new_scan().plan()
+ assert sum(file.row_count for split in original_plan.splits() for file in
split.files) == len(values)
+ catalog.alter_table('default.t', [SchemaChange.update_column_type(path,
AtomicType(target))], False)
+ table = catalog.get_table('default.t')
+ planning = table.copy({'scan.native-plan.enabled': 'false',
'read.native.enabled': 'false'})
+ splits = planning.new_read_builder().new_scan().plan().splits()
+ assert splits
+ for split in splits:
+ for meta in split.files:
+ location = meta.file_path
+ if location.startswith('file:'):
+ location = unquote(urlparse(location).path)
+ data = pq.read_table(location)
+ index = data.schema.get_field_index('v')
+ field = data.schema.field(index)
+ data = data.set_column(index, pa.field(
+ field.name, physical, nullable=field.nullable,
metadata=field.metadata),
+ data.column(index).cast(physical))
+ pq.write_table(data, location, row_group_size=2)
+ meta.file_size = os.stat(location).st_size
+ if not pa.types.is_map(physical):
+ assert pq.read_schema(location).field('v').type == physical
+ # PyArrow restores LargeList at the top level and inside ROW,
+ # while its Parquet reader normalizes MAP values to regular List.
+ for native in (False, True):
+ reader = table.copy({'read.native.enabled':
str(native).lower()}).new_read_builder().new_read()
+ guard = patch('pypaimon.read.table_read.TableRead._create_split_read',
+ side_effect=AssertionError('Native read fell back')) if
native else ExitStack()
+ with guard:
+ result = reader.to_arrow(splits).sort_by([('id', 'ascending')])
+ # Java's constructed casts use STRING for children and ignore the
+ # outer CHAR length. NULL containers and NULL elements stay distinct.
+ assert result.column('v').to_pylist() == expected
diff --git a/paimon-python/pypaimon/tests/native_read_type_test.py
b/paimon-python/pypaimon/tests/native_read_type_test.py
index f15f1c0f43..a1625f85d9 100644
--- a/paimon-python/pypaimon/tests/native_read_type_test.py
+++ b/paimon-python/pypaimon/tests/native_read_type_test.py
@@ -396,6 +396,27 @@ def
test_whole_parent_and_child_apply_authorization_before_extraction(
assert sorted(row[1] for row in rows if row[1] is not None) == [20]
[email protected]('native', [False, True])
+def test_map_projection_preserves_masked_key_nullability(tmp_path, native):
+ from pypaimon.catalog.filesystem_catalog import FileSystemCatalog
+ from pypaimon.catalog.table_query_auth import TableQueryAuthResult
+
+ table, _ = _table(tmp_path, 'pk', True)
+ table = table.copy({'read.native.enabled': str(native).lower(),
'query-auth.enabled': 'true'})
+ builder = table.new_read_builder().with_projection(['attrs',
"attrs['overflow']", 'id'])
+ auth = TableQueryAuthResult(None, {'id': json.dumps({'name': 'NULL'})})
+ # A logical NOT NULL key becomes nullable after masking. Test the captured
+ # batch schema independently of the core's conservative read nullability.
+ with patch.object(FileSystemCatalog, 'auth_table_query',
return_value=auth), \
+ patch('pypaimon.read.table_read.read_arrow_schema',
PyarrowFieldParser.from_paimon_schema):
+ splits = builder.new_scan().plan().splits()
+ result = builder.new_read().to_arrow_batch_reader(splits).read_all()
+ assert result.num_rows == 3
+ assert result.schema.field('id').nullable
+ assert result.column('id').to_pylist() == [None] * 3
+ assert sorted(value for value in
result.column('attrs_overflow').to_pylist() if value is not None) == [20]
+
+
def
test_clipped_variant_reads_only_selected_paths_for_all_json_kinds(tmp_path):
table, _ = _table(tmp_path, 'append', False, variant_shredded=True)
schema = PyarrowFieldParser.from_paimon_schema(table.fields)
diff --git a/paimon-python/pypaimon/tests/native_schema_cast_test.py
b/paimon-python/pypaimon/tests/native_schema_cast_test.py
new file mode 100644
index 0000000000..1a1e90adae
--- /dev/null
+++ b/paimon-python/pypaimon/tests/native_schema_cast_test.py
@@ -0,0 +1,243 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Python schema casts transport Arrow arrays into the core Java executors."""
+
+from decimal import Decimal
+from unittest.mock import patch
+
+import pyarrow as pa
+import pytest
+
+from pypaimon.casting.schema_evolution import cast_array_for_schema_evolution
+from pypaimon.schema.data_types import (ArrayType, AtomicType, DataField,
DataTypeParser,
+ MapType, PyarrowFieldParser, RowType)
+from pypaimon.tests import native_schema_evolution_cast_test
+
+rest_catalog = native_schema_evolution_cast_test.rest_catalog
+pytestmark = native_schema_evolution_cast_test.pytestmark
+
+_CASES = [
+ ('STRING', 'FLOAT', [' 1.25 ', '0x1.8p2', '1.25f', None], [1.25, 6.0,
1.25, None]),
+ ('STRING', 'DOUBLE', [' 1.25 ', '0x1.8p2', '1.25D', None], [1.25, 6.0,
1.25, None]),
+ ('FLOAT', 'STRING', [0.0001, 1e23, 16777216, None], ['1.0E-4', '1.0E23',
'1.6777216E7', None]),
+ ('DOUBLE', 'STRING', [0.0001, 1e23, None], ['1.0E-4',
'9.999999999999999E22', None]),
+ ('STRING', 'CHAR(3)', ['πδΈ', 'a', 'abcdef', None], ['πδΈ ', 'a ', 'abc',
None]),
+ ('STRING', 'VARCHAR(2)', ['πδΈx', 'a', None], ['πδΈ', 'a', None]),
+ ('STRING', 'BINARY(3)', ['ab', 'abcd', None], [b'ab\x00', b'abc', None]),
+ ('STRING', 'VARBINARY(1)', ['π', None], [b'\xf0', None]),
+ ('INT', 'DECIMAL(4,2)', [1, 100, None], [Decimal('1.00'), None, None]),
+ ('STRING', 'DATE', ['42', '2001-02-28', None], [42, 11381, None]),
+ ('STRING', 'TIMESTAMP(1)', ['1970-01-01 00:00:00.123456', None], [100,
None]),
+ ('TIMESTAMP(3)', 'TIMESTAMP(1)', [-(1 << 63), -1, 123, None],
+ [9223372036854775716, -100, 100, None]),
+ ('INT', 'TIMESTAMP(6)', [1, -1, None], [1000000, -1000000, None]),
+ ('INTEGER', 'BIGINT', [1, -1, None], [1, -1, None]),
+ ('DECIMAL', 'STRING', [Decimal('12'), None], ['12', None]),
+ ('TIME', 'STRING', [0, 1000, None], ['00:00:00', '00:00:01', None]),
+ ('TIMESTAMP', 'STRING', [0, 100000, None],
+ ['1970-01-01 00:00:00.000000', '1970-01-01 00:00:00.100000', None]),
+ ('TIMESTAMP_LTZ', 'TIMESTAMP_LTZ(3)', [0, 100000, None], [0, 100, None]),
+]
+
+
+def _type(name):
+ logical = DataTypeParser.parse_data_type(name)
+ return logical, PyarrowFieldParser.from_paimon_type(logical)
+
+
[email protected]('source,target,values,expected', _CASES,
+ ids=lambda value: value if isinstance(value, str)
else None)
[email protected]('chunked', [False, True])
+def test_python_schema_casts_use_rust(source, target, values, expected,
chunked):
+ source_type, source_arrow = _type(source)
+ target_type, target_arrow = _type(target)
+ array = pa.array(values, type=source_arrow)
+ if chunked:
+ array = pa.chunked_array([array.slice(0, 1), array.slice(1),
pa.array([], type=source_arrow)])
+ with patch('pypaimon.casting.schema_evolution.cast_decimal',
side_effect=AssertionError('Python scalar cast')), \
+ patch('pypaimon.casting.schema_evolution.cast_temporal',
+ side_effect=AssertionError('Python temporal cast')):
+ result = cast_array_for_schema_evolution(array, target_arrow,
source_type, target_type)
+ if chunked:
+ assert isinstance(result, pa.ChunkedArray)
+ result = result.combine_chunks()
+ assert result.equals(pa.array(expected, type=target_arrow))
+
+
[email protected]('mode', ['append', 'pk', 'de'])
[email protected]('source,target,values,expected', _CASES,
+ ids=lambda value: value if isinstance(value, str)
else None)
+def test_existing_scalar_evolution_uses_native_rest_reader(rest_catalog, mode,
source, target, values, expected):
+ if mode == 'pk' and source == 'TIMESTAMP(3)' and target == 'TIMESTAMP(1)':
+ pytest.skip('The existing Python PK writer cannot materialize the
extreme timestamp input')
+ catalog, _ = rest_catalog
+ source_type, source_arrow = _type(source)
+ _, target_arrow = _type(target)
+ native_schema_evolution_cast_test._table(catalog, mode, source_arrow,
values, source_type)
+ from pypaimon.schema.schema_change import SchemaChange
+ catalog.alter_table('default.t', [SchemaChange.update_column_type('v',
AtomicType(target))], False)
+ result =
native_schema_evolution_cast_test._read(catalog.get_table('default.t'), True)
+ assert result.column('v').combine_chunks().equals(pa.array(expected,
type=target_arrow))
+
+
[email protected]('target', ['INT', 'FLOAT', 'DOUBLE', 'DATE',
'TIMESTAMP(3)'])
+def test_invalid_scalar_text_is_an_error_instead_of_null(target):
+ logical, arrow = _type(target)
+ with pytest.raises((ValueError, RuntimeError)):
+ cast_array_for_schema_evolution(pa.array(['bad']), arrow,
AtomicType('STRING'), logical)
+
+
+def test_empty_chunked_input_retains_the_requested_type():
+ result = cast_array_for_schema_evolution(
+ pa.chunked_array([], type=pa.string()), pa.int32(),
AtomicType('STRING'), AtomicType('INT'))
+ assert result.type == pa.int32() and len(result) == 0
+
+
+def test_time_precision_changes_are_not_executable():
+ from pypaimon.casting.data_type_casts import can_execute_cast
+ assert not can_execute_cast(AtomicType('TIME(3)'), AtomicType('TIME(1)'))
+ with pytest.raises(NotImplementedError):
+ cast_array_for_schema_evolution(
+ pa.array([], type=pa.time32('ms')), pa.time32('ms'),
AtomicType('TIME(3)'), AtomicType('TIME(1)'))
+
+
[email protected]('source,target', [('TIME', 'TIME(0)'), ('TIME(3)',
'TIME( 3 )'), ('TIME(3)', 'TIME(03)')])
+def test_time_equivalent_precision_spellings_are_identity(source, target):
+ from pypaimon.casting.data_type_casts import can_execute_cast
+ assert can_execute_cast(AtomicType(source), AtomicType(target))
+ data = pa.array([100, None], type=pa.time32('ms'))
+ assert cast_array_for_schema_evolution(data, data.type,
AtomicType(source), AtomicType(target)).equals(data)
+
+
+def _decimal_container(kind, precision):
+ leaf = AtomicType('DECIMAL({})'.format(precision), False)
+ if kind == 'array':
+ return ArrayType(True, leaf), ['v', 'element']
+ if kind == 'row':
+ return RowType(True, [DataField(2, 'd', leaf)]), ['v', 'd']
+ if kind == 'map':
+ return MapType(True, AtomicType('STRING', False), leaf), ['v', 'value']
+ return leaf, 'v'
+
+
+_DECIMAL_CONTAINERS = [
+ ('leaf', [Decimal('9.995'), Decimal('1.000')], [None, Decimal('1.00')]),
+ ('array', [[Decimal('9.995'), Decimal('1.000')], None, []], [[None,
Decimal('1.00')], None, []]),
+ ('row', [{'d': Decimal('9.995')}, None, {'d': Decimal('1.000')}],
+ [{'d': None}, None, {'d': Decimal('1.00')}]),
+ ('map', [[('a', Decimal('9.995')), ('b', Decimal('1.000'))], None, []],
+ [[('a', None), ('b', Decimal('1.00'))], None, []]),
+]
+
+
[email protected]('kind,values,expected', _DECIMAL_CONTAINERS)
[email protected]('chunked', [False, True])
+def test_not_null_casts_can_return_java_runtime_nulls(kind, values, expected,
chunked):
+ source, _ = _decimal_container(kind, '6,3')
+ target, _ = _decimal_container(kind, '3,2')
+ data = pa.array(values, type=PyarrowFieldParser.from_paimon_type(source))
+ if chunked:
+ data = pa.chunked_array([data.slice(0, 1), data.slice(1)])
+ result = cast_array_for_schema_evolution(data,
PyarrowFieldParser.from_paimon_type(target), source, target)
+ assert result.to_pylist() == expected
+ result.validate(full=True)
+
+
[email protected]('mode', ['append', 'pk', 'de'])
[email protected]('kind,values,expected', _DECIMAL_CONTAINERS)
+def test_not_null_decimal_evolution_uses_native_rest_reader(rest_catalog,
mode, kind, values, expected):
+ catalog, _ = rest_catalog
+ source, path = _decimal_container(kind, '6,3')
+ native_schema_evolution_cast_test._table(
+ catalog, mode, PyarrowFieldParser.from_paimon_type(source), values,
source)
+ from pypaimon.schema.schema_change import SchemaChange
+ catalog.alter_table('default.t', [SchemaChange.update_column_type(path,
AtomicType('DECIMAL(3,2)'), True)], False)
+ table = catalog.get_table('default.t')
+ for native in [False, True]:
+ result = native_schema_evolution_cast_test._read(table, native)
+ assert result.column('v').to_pylist() == expected
+ # Read metadata permits cast-produced NULLs; persisted constraints are
intact.
+ logical = catalog.get_table('default.t').fields[1].type
+ if kind == 'array':
+ logical = logical.element
+ elif kind == 'row':
+ logical = logical.fields[0].type
+ elif kind == 'map':
+ logical = logical.value
+ assert not logical.nullable
+
+
[email protected]('target,expected', [
+ ('DECIMAL(10,2)', [Decimal('1.24'), None]), ('INT', [1, None]), ('STRING',
['1.235', None]),
+])
+def test_decimal256_physical_storage_uses_the_logical_decimal_type(target,
expected):
+ logical, arrow = _type(target)
+ data = pa.array([Decimal('1.235'), None], type=pa.decimal256(20, 3))
+ result = cast_array_for_schema_evolution(data, arrow,
AtomicType('DECIMAL(20,3)'), logical)
+ assert result.equals(pa.array(expected, type=arrow))
+
+
[email protected]('chunked', [False, True])
+def test_large_list_rows_align_by_field_id_in_the_shared_core(chunked):
+ source_row = RowType(True, [DataField(1, 'old', AtomicType('STRING')),
+ DataField(2, 'dropped', AtomicType('INT'))])
+ target_row = RowType(True, [DataField(2, 'dropped', AtomicType('BIGINT')),
+ DataField(1, 'renamed', AtomicType('INT')),
+ DataField(3, 'added', AtomicType('INT'))])
+ source, target = ArrayType(True, source_row), ArrayType(True, target_row)
+ physical = pa.large_list(pa.struct([('old', pa.large_string()),
('dropped', pa.int32())]))
+ data = pa.array([[{'old': 'bad', 'dropped': 0}],
+ [{'old': '1.9', 'dropped': 7}, None], None, [],
+ [{'old': 'bad', 'dropped': 0}]], type=physical).slice(1,
3)
+ if chunked:
+ data = pa.chunked_array([data.slice(0, 1), data.slice(1)])
+ result = cast_array_for_schema_evolution(data,
PyarrowFieldParser.from_paimon_type(target), source, target)
+ assert result.to_pylist() == [[{'dropped': 7, 'renamed': 1, 'added':
None}, None], None, []]
+
+
[email protected]('native_write', [False, True])
[email protected]('kind,values,invalid', [
+ ('array', [[Decimal('1.000')], None, []], [[None]]),
+ ('row', [{'d': Decimal('1.000')}, None], [{'d': None}]),
+ ('map', [[('a', Decimal('1.000'))], None, []], [[('a', None)]]),
+])
+def test_read_results_write_back_with_not_null_constraints(rest_catalog,
native_write, kind, values, invalid):
+ catalog, _ = rest_catalog
+ logical, _ = _decimal_container(kind, '6,3')
+ native_schema_evolution_cast_test._table(
+ catalog, 'append', PyarrowFieldParser.from_paimon_type(logical),
values, logical)
+ table = catalog.get_table('default.t')
+ data = native_schema_evolution_cast_test._read(table, True)
+ table = table.copy({'write.native.enabled': str(native_write).lower()})
+ builder = table.new_batch_write_builder()
+ writer, commit = builder.new_write(), builder.new_commit()
+ from pypaimon.write.native_write import NativeTableWrite
+ assert isinstance(writer, NativeTableWrite) == native_write
+ try:
+ writer.write_arrow(data)
+ commit.commit(writer.prepare_commit())
+ bad = pa.Table.from_arrays(
+ [pa.array([100], type=pa.int64()), pa.array(invalid,
type=data.schema.field('v').type)],
+ schema=data.schema)
+ with pytest.raises(ValueError):
+ writer.write_arrow(bad)
+ finally:
+ writer.close()
+ commit.close()
+ output =
native_schema_evolution_cast_test._read(catalog.get_table('default.t'), True)
+ assert output.num_rows == 2 * len(values)
diff --git
a/paimon-python/pypaimon/tests/py36/filter_record_batch_reader_test.py
b/paimon-python/pypaimon/tests/py36/filter_record_batch_reader_test.py
new file mode 100644
index 0000000000..fbeb43a94c
--- /dev/null
+++ b/paimon-python/pypaimon/tests/py36/filter_record_batch_reader_test.py
@@ -0,0 +1,64 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import unittest
+
+import pyarrow as pa
+
+from pypaimon.common.predicate import Predicate
+from pypaimon.read.reader.filter_record_batch_reader import
FilterRecordBatchReader
+from pypaimon.read.reader.iface.record_batch_reader import RecordBatchReader
+from pypaimon.schema.data_types import AtomicType, DataField
+
+
+class _Batches(RecordBatchReader):
+ def __init__(self, batches):
+ self.batches = iter(batches)
+
+ def read_arrow_batch(self, start_idx=None, end_idx=None):
+ return next(self.batches, None)
+
+ def close(self):
+ pass
+
+
+class FilterRecordBatchReaderTest(unittest.TestCase):
+ def test_row_filter_uses_logical_field_order_and_preserves_metadata(self):
+ schema = pa.schema([
+ ('_row_kind', pa.string()), ('id', pa.int64()), ('behavior',
pa.string()),
+ ], metadata={b'source': b'row-filter'})
+ batch = pa.RecordBatch.from_arrays([
+ pa.array(['+I', '-U', '+U', '-D']), pa.array([1, 2, 3, 4]),
+ pa.array(['apple', None, 'pear', 'apricot']),
+ ], schema=schema)
+ # The empty first result must not terminate the reader.
+ source = _Batches([batch.slice(2, 1), batch])
+ predicate = Predicate('or', None, None, [
+ Predicate('startsWith', 0, 'behavior', ['a']),
+ Predicate('isNull', 0, 'behavior', []),
+ ])
+ fields = [DataField(0, 'behavior', AtomicType('STRING')), DataField(1,
'id', AtomicType('BIGINT'))]
+ reader = FilterRecordBatchReader(source, predicate,
schema_fields=fields)
+ try:
+ result = reader.read_arrow_batch()
+ self.assertTrue(result.schema.equals(schema, check_metadata=True))
+ self.assertEqual(result.column(0).to_pylist(), ['+I', '-U', '-D'])
+ self.assertEqual(result.column(1).to_pylist(), [1, 2, 4])
+ self.assertEqual(result.column(2).to_pylist(), ['apple', None,
'apricot'])
+ self.assertIsNone(reader.read_arrow_batch())
+ finally:
+ reader.close()
diff --git a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
index 363986c77d..f6ce193f57 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_cast_test.py
@@ -28,6 +28,13 @@ from pypaimon.casting.schema_evolution import
cast_array_for_schema_evolution
from pypaimon.schema.data_types import AtomicType
[email protected](autouse=True)
+def python_fallback_casts(monkeypatch):
+ # These tests exercise libc/Python fallbacks, including mocked localtime
+ # and changes to TZ. Native casts have separate process-zone regressions.
+
monkeypatch.setattr('pypaimon.casting.schema_evolution._native_schema_cast',
lambda *args: None)
+
+
@pytest.mark.parametrize('chunked', [False, True])
def test_decimal_evolution_uses_supported_jvm_floating_decimal_digits(chunked):
# Java 8/11/17 retain exact digits in this integral fast path. JDK 19+
diff --git a/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
b/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
index 899634fafe..8bbbd072d0 100644
--- a/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
+++ b/paimon-python/pypaimon/tests/schema_evolution_nested_read_test.py
@@ -777,11 +777,17 @@ class SupportsCastTest(unittest.TestCase):
# A numeric -> decimal cast has a PyArrow kernel but is only executable
# when the target precision can hold the source's range at the target
# scale (INT needs >= 12 at scale 2, BIGINT >= 21). An empty-array
probe
- # misses this; can_execute_cast must reject the too-small targets so
the
- # read path does not later fail with ArrowInvalid.
+ # misses this. The Python fallback must reject those targets; the core
+ # executor supports them and returns NULL for per-value Java overflow.
+ try:
+ from pypaimon_rust.data import cast_array_for_schema_evolution
+ except ImportError:
+ native = False
+ else:
+ native = callable(cast_array_for_schema_evolution)
for src, dst in [('INT', 'DECIMAL(10, 2)'), ('BIGINT', 'DECIMAL(10,
2)'),
('BIGINT', 'DECIMAL(20, 2)')]:
- self.assertFalse(can_execute_cast(AtomicType(src),
AtomicType(dst)),
+ self.assertEqual(can_execute_cast(AtomicType(src),
AtomicType(dst)), native,
'{} -> {}'.format(src, dst))
for src, dst in [('INT', 'DECIMAL(12, 2)'), ('BIGINT', 'DECIMAL(21,
2)'),
('INT', 'BIGINT'), ('DOUBLE', 'INT'), ('INT',
'STRING')]:
diff --git a/paimon-python/pypaimon/write/native_write.py
b/paimon-python/pypaimon/write/native_write.py
index 0897ca8e95..e737c62497 100644
--- a/paimon-python/pypaimon/write/native_write.py
+++ b/paimon-python/pypaimon/write/native_write.py
@@ -21,7 +21,8 @@ from importlib import import_module
import pyarrow as pa
from pypaimon.common.options.core_options import CoreOptions
-from pypaimon.schema.arrow_schema import arrow_schemas_compatible,
normalize_arrow_strings
+from pypaimon.schema.arrow_schema import (
+ arrow_schemas_compatible, normalize_arrow_strings,
normalize_arrow_write_nullability)
from pypaimon.schema.data_types import PyarrowFieldParser, is_blob_file_field
from pypaimon.table.bucket_mode import BucketMode
from pypaimon.write.native_commit import (
@@ -226,6 +227,9 @@ class NativeTableWrite:
self._native_writer.write_arrow(data, bucket)
def _prepare_native_arrow_data(self, data):
+ data = normalize_arrow_strings(data)
+ target = self._table_schema if data.schema.names ==
self._table_schema.names else self._schema
+ data = normalize_arrow_write_nullability(data, target)
if self._write_cols is not None and arrow_schemas_compatible(
data.schema, self._table_schema,
check_top_level_nullability=False,
allow_binary_compatibility=True):
@@ -239,7 +243,7 @@ class NativeTableWrite:
"Input schema isn't consistent with table schema and write
cols. "
f"Input schema is: {data.schema} Table schema is:
{self._schema} "
f"Write cols is: {self._write_cols}")
- return normalize_arrow_strings(data)
+ return data
def write_pandas(self, dataframe):
if self._python_writer is not None:
diff --git a/paimon-python/pypaimon/write/table_write.py
b/paimon-python/pypaimon/write/table_write.py
index 44909ea45a..c78ea60801 100644
--- a/paimon-python/pypaimon/write/table_write.py
+++ b/paimon-python/pypaimon/write/table_write.py
@@ -20,7 +20,8 @@ from typing import TYPE_CHECKING, Any, Dict, List, Optional
import pyarrow as pa
from pypaimon.common.options.core_options import ChangelogProducer
-from pypaimon.schema.arrow_schema import arrow_schemas_compatible,
normalize_arrow_strings
+from pypaimon.schema.arrow_schema import (
+ arrow_schemas_compatible, normalize_arrow_strings,
normalize_arrow_write_nullability)
from pypaimon.schema.data_types import PyarrowFieldParser
from pypaimon.snapshot.snapshot import BATCH_COMMIT_IDENTIFIER
from pypaimon.table.row.blob import BlobConsumer
@@ -72,7 +73,7 @@ class TableWrite:
def write_arrow(self, data, bucket: Optional[int] = None):
"""Write Arrow data, optionally to a bucket assigned upstream."""
- self._validate_pyarrow_schema(data.schema)
+ data = self._prepare_arrow_schema(data)
if isinstance(data, pa.RecordBatch):
return self.write_arrow_batch(data, bucket)
for batch in data.to_batches():
@@ -308,8 +309,7 @@ class TableWrite:
release()
def _prepare_arrow_data(self, data):
- self._validate_pyarrow_schema(data.schema)
- data = normalize_arrow_strings(data)
+ data = self._prepare_arrow_schema(data)
# Schema-only writers can normalize input without table options.
table = getattr(self, 'table', None)
if table is None:
@@ -318,6 +318,18 @@ class TableWrite:
from pypaimon.write.row_kind import filter_write_batch
return filter_write_batch(table, data)
+ def _prepare_arrow_schema(self, data):
+ data = normalize_arrow_strings(data)
+ target = self.table_pyarrow_schema
+ if data.schema.names != target.names and
self.file_store_write.write_cols is not None:
+ target =
self._write_cols_pyarrow_schema(self.file_store_write.write_cols)
+ validated = normalize_arrow_write_nullability(data, target)
+ self._validate_pyarrow_schema(validated.schema)
+ # Core checks only visible NOT NULL children. PyArrow's Parquet writer
+ # also rejects hidden NULL children under NULL ROW parents when their
+ # Arrow field is nonnullable, so retain the input's transport metadata.
+ return data
+
def _normalize_arrow_write_schema(self, data):
# Compatible input batches may differ in top-level nullability or
# metadata. Buffered writes and row-to-Arrow conversion must use the
