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 52c0b51c2c [python][daft] Fix native reads after schema evolution
(#8704)
52c0b51c2c is described below
commit 52c0b51c2c6e0e9c09ce907314ee9d966a2947f4
Author: QuakeWang <[email protected]>
AuthorDate: Sat Jul 18 12:05:37 2026 +0800
[python][daft] Fix native reads after schema evolution (#8704)
---
paimon-python/pypaimon/daft/daft_datasource.py | 218 +++++++++++-
.../tests/daft/daft_blob_native_read_test.py | 35 ++
.../pypaimon/tests/daft/daft_datasource_test.py | 335 ++++++++++++++++++-
.../pypaimon/tests/daft/daft_explain_test.py | 110 ++++++
.../pypaimon/tests/daft/daft_integration_test.py | 368 ++++++++++++++++++++-
5 files changed, 1058 insertions(+), 8 deletions(-)
diff --git a/paimon-python/pypaimon/daft/daft_datasource.py
b/paimon-python/pypaimon/daft/daft_datasource.py
index 10c2a51cf9..dff73f02bc 100644
--- a/paimon-python/pypaimon/daft/daft_datasource.py
+++ b/paimon-python/pypaimon/daft/daft_datasource.py
@@ -40,7 +40,17 @@ from pypaimon.daft.daft_explain import (
)
from pypaimon.daft.daft_predicate_visitor import convert_filters_to_paimon
from pypaimon.read.query_auth_split import QueryAuthSplit
-from pypaimon.schema.data_types import is_array_blob_type, is_blob_type
+from pypaimon.schema.data_types import (
+ ArrayType,
+ AtomicType,
+ DataField,
+ DataType,
+ MapType,
+ RowType,
+ VectorType,
+ is_array_blob_type,
+ is_blob_type,
+)
if TYPE_CHECKING:
from collections.abc import AsyncIterator
@@ -62,6 +72,117 @@ PAIMON_FILE_FORMAT_AVRO = "avro"
_PaimonIdentifier = tuple[str, str, str | None]
+# Daft's Parquet reader applies casts independently of PyPaimon. Keep this
+# list limited to promotions whose results are covered by end-to-end parity
+# tests; logical schema-change support alone is not sufficient.
+_NATIVE_READ_ATOMIC_PROMOTIONS = frozenset({("INT", "BIGINT")})
+
+
+def _promote_time32_type_for_daft(data_type: pa.DataType) -> pa.DataType:
+ """Use Daft's supported time representation without changing values."""
+ if pa.types.is_time32(data_type):
+ return pa.time64("us")
+ if pa.types.is_struct(data_type):
+ fields = [_promote_time32_field_for_daft(field) for field in data_type]
+ return data_type if fields == list(data_type) else pa.struct(fields)
+ if pa.types.is_list(data_type):
+ value_field = _promote_time32_field_for_daft(data_type.value_field)
+ return data_type if value_field == data_type.value_field else
pa.list_(value_field)
+ if pa.types.is_map(data_type):
+ key_type = _promote_time32_type_for_daft(data_type.key_type)
+ item_type = _promote_time32_type_for_daft(data_type.item_type)
+ return (
+ data_type
+ if key_type == data_type.key_type and item_type ==
data_type.item_type
+ else pa.map_(key_type, item_type,
keys_sorted=data_type.keys_sorted)
+ )
+ return data_type
+
+
+def _promote_time32_field_for_daft(field: pa.Field) -> pa.Field:
+ data_type = _promote_time32_type_for_daft(field.type)
+ if data_type == field.type:
+ return field
+ return pa.field(
+ field.name,
+ data_type,
+ nullable=field.nullable,
+ metadata=field.metadata,
+ )
+
+
+def _promote_time32_schema_for_daft(schema: pa.Schema) -> pa.Schema:
+ fields = [_promote_time32_field_for_daft(field) for field in schema]
+ if fields == list(schema):
+ return schema
+ return pa.schema(fields, metadata=schema.metadata)
+
+
+def _promote_time32_batch_for_daft(batch: pa.RecordBatch) -> pa.RecordBatch:
+ schema = _promote_time32_schema_for_daft(batch.schema)
+ return batch if schema == batch.schema else batch.cast(schema, safe=False)
+
+
+def _native_read_fields_compatible(
+ file_fields: list[DataField],
+ current_fields: list[DataField],
+) -> bool:
+ """Whether Daft can align the selected current fields by physical name."""
+ file_fields_by_id = {field.id: field for field in file_fields}
+ file_fields_by_name = {field.name: field for field in file_fields}
+
+ for current_field in current_fields:
+ file_field = file_fields_by_id.get(current_field.id)
+ if file_field is None:
+ # A missing nullable field is a later addition and Daft fills it
+ # with NULL. The same physical name under another id is instead a
+ # drop-then-readd and must be resolved by PyPaimon.
+ if (
+ current_field.name in file_fields_by_name
+ or not current_field.type.nullable
+ ):
+ return False
+ continue
+ if file_field.name != current_field.name:
+ return False
+ if not _native_read_types_compatible(file_field.type,
current_field.type):
+ return False
+ return True
+
+
+def _native_read_types_compatible(
+ file_type: DataType,
+ current_type: DataType,
+) -> bool:
+ if file_type.nullable and not current_type.nullable:
+ return False
+ if type(file_type) is not type(current_type):
+ return False
+ if isinstance(file_type, RowType) and isinstance(current_type, RowType):
+ return _native_read_fields_compatible(file_type.fields,
current_type.fields)
+ if isinstance(file_type, ArrayType) and isinstance(current_type,
ArrayType):
+ return _native_read_types_compatible(file_type.element,
current_type.element)
+ if isinstance(file_type, VectorType) and isinstance(current_type,
VectorType):
+ return (
+ file_type.length == current_type.length
+ and _native_read_types_compatible(file_type.element,
current_type.element)
+ )
+ if isinstance(file_type, MapType) and isinstance(current_type, MapType):
+ return (
+ _native_read_types_compatible(file_type.key, current_type.key)
+ and _native_read_types_compatible(file_type.value,
current_type.value)
+ )
+ if not isinstance(file_type, AtomicType):
+ return False
+ file_type_name = file_type.type.upper()
+ current_type_name = current_type.type.upper()
+ if file_type_name == "TIME" or file_type_name.startswith("TIME("):
+ return False
+ return (
+ file_type_name == current_type_name
+ or (file_type_name, current_type_name) in
_NATIVE_READ_ATOMIC_PROMOTIONS
+ )
+
@dataclass(frozen=True, slots=True)
class _ReadPushdownState:
@@ -235,6 +356,7 @@ class _PaimonPKSplitTask(DataSourceTask):
blob_io_config_bytes,
self._array_blob_column_names,
)
+ batch = _promote_time32_batch_for_daft(batch)
rb = RecordBatch.from_arrow_record_batches([batch], batch.schema)
if has_blob_columns:
rb = _cast_blob_columns_to_file(
@@ -473,7 +595,9 @@ class PaimonDataSource(DataSource):
from pypaimon.schema.data_types import PyarrowFieldParser
- pa_schema = PyarrowFieldParser.from_paimon_schema(table.fields)
+ pa_schema = _promote_time32_schema_for_daft(
+ PyarrowFieldParser.from_paimon_schema(table.fields)
+ )
self._scalar_blob_column_names = {
field.name for field in table.fields if is_blob_type(field.type)
@@ -513,7 +637,12 @@ class PaimonDataSource(DataSource):
self._is_parquet = self._file_format == PAIMON_FILE_FORMAT_PARQUET
self._partition_field_arrow_types: dict[str, pa.DataType] = (
- {f.name: PyarrowFieldParser.from_paimon_type(f.type) for f in
table.partition_keys_fields}
+ {
+ f.name: _promote_time32_type_for_daft(
+ PyarrowFieldParser.from_paimon_type(f.type)
+ )
+ for f in table.partition_keys_fields
+ }
if table.partition_keys
else {}
)
@@ -595,16 +724,18 @@ class PaimonDataSource(DataSource):
plan = read_builder.new_scan().plan()
pv_cache: dict[tuple[tuple[str, Any], ...], RecordBatch | None] = {}
+ schema_incompatibility_cache: dict[int, bool] = {}
for split in plan.splits():
if self._partition_filter_skips_split(split, pushdowns, pv_cache):
continue
has_deletion_vectors = self._split_has_deletion_vectors(split)
+ has_auth = self._split_has_auth(split)
routing = self._reader_routing(
raw_convertible=split.raw_convertible,
has_deletion_vectors=has_deletion_vectors,
- has_auth=self._split_has_auth(split),
+ has_auth=has_auth,
)
native_files = (
@@ -614,6 +745,19 @@ class PaimonDataSource(DataSource):
split.files, read_pushdowns.task_columns,
has_deletion_vectors
)
)
+ if native_files is not None and self._has_incompatible_file_schema(
+ read_table,
+ [data_file.schema_id for data_file in native_files],
+ read_pushdowns.task_columns,
+ schema_incompatibility_cache,
+ ):
+ native_files = None
+ routing = self._reader_routing(
+ raw_convertible=split.raw_convertible,
+ has_deletion_vectors=has_deletion_vectors,
+ has_auth=has_auth,
+ has_incompatible_schema=True,
+ )
if native_files is not None:
task_schema = (
@@ -673,6 +817,7 @@ class PaimonDataSource(DataSource):
fallback_reasons: dict[str, int] = {}
explained_splits: list[PaimonReaderSplitExplain] | None = [] if
verbose else None
pv_cache: dict[tuple[tuple[str, Any], ...], RecordBatch | None] = {}
+ schema_incompatibility_cache: dict[int, bool] = {}
for split in split_details:
if self._partition_filter_skips_explain_split(split, pushdowns,
pv_cache):
@@ -692,6 +837,27 @@ class PaimonDataSource(DataSource):
split.has_deletion_vectors,
)
)
+ candidate_files = (
+ getattr(split, "data_files", None)
+ if routing.use_native_reader
+ else blob_native_files
+ )
+ candidate_schema_ids = [
+ data_file.schema_id for data_file in candidate_files or []
+ ]
+ if candidate_schema_ids and self._has_incompatible_file_schema(
+ read_table,
+ candidate_schema_ids,
+ read_pushdowns.task_columns,
+ schema_incompatibility_cache,
+ ):
+ blob_native_files = None
+ routing = self._reader_routing(
+ raw_convertible=split.raw_convertible,
+ has_deletion_vectors=split.has_deletion_vectors,
+ has_auth=paimon_scan.has_auth,
+ has_incompatible_schema=True,
+ )
# For a blob-native split only the covering parquet files are read
# natively; report their counts so the verbose per-split detail
@@ -767,6 +933,7 @@ class PaimonDataSource(DataSource):
raw_convertible: bool,
has_deletion_vectors: bool,
has_auth: bool = False,
+ has_incompatible_schema: bool = False,
) -> _ReaderRouting:
can_use_native_reader = (
self._is_parquet
@@ -774,20 +941,27 @@ class PaimonDataSource(DataSource):
and raw_convertible
and not has_deletion_vectors
and not has_auth
+ and not has_incompatible_schema
)
if can_use_native_reader:
return _ReaderRouting(READER_MODE_NATIVE_PARQUET, None)
if not self._is_parquet:
reason = "non-parquet format"
+ elif has_incompatible_schema:
+ reason = "schema evolution requires PyPaimon normalization"
elif self._has_blob_columns:
reason = "blob columns present"
elif has_auth:
reason = "query auth active"
elif has_deletion_vectors:
reason = "deletion vectors present"
- elif self._table.is_primary_key_table:
- reason = "LSM merge required"
+ elif not raw_convertible:
+ reason = (
+ "LSM merge required"
+ if self._table.is_primary_key_table
+ else "data-evolution merge required"
+ )
else:
reason = "data-evolution merge required"
return _ReaderRouting(READER_MODE_PYPAIMON_FALLBACK, reason)
@@ -815,6 +989,38 @@ class PaimonDataSource(DataSource):
files, task_columns, blob_column_names, self._table.partition_keys
)
+ @staticmethod
+ def _has_incompatible_file_schema(
+ table: FileStoreTable,
+ schema_ids: list[int],
+ task_columns: list[str] | None,
+ cache: dict[int, bool],
+ ) -> bool:
+ current_schema = table.table_schema
+ if task_columns is None:
+ current_fields = current_schema.fields
+ else:
+ task_column_set = set(task_columns)
+ current_fields = [
+ field
+ for field in current_schema.fields
+ if field.name in task_column_set
+ ]
+ for schema_id in schema_ids:
+ if schema_id not in cache:
+ file_schema = (
+ current_schema
+ if schema_id == current_schema.id
+ else table.schema_manager.get_schema(schema_id)
+ )
+ cache[schema_id] = not _native_read_fields_compatible(
+ file_schema.fields,
+ current_fields,
+ )
+ if cache[schema_id]:
+ return True
+ return False
+
@staticmethod
def _split_has_deletion_vectors(split: Split) -> bool:
deletion_files = getattr(split, "data_deletion_files", None)
diff --git a/paimon-python/pypaimon/tests/daft/daft_blob_native_read_test.py
b/paimon-python/pypaimon/tests/daft/daft_blob_native_read_test.py
index 202361e3ab..ba6a762915 100644
--- a/paimon-python/pypaimon/tests/daft/daft_blob_native_read_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_blob_native_read_test.py
@@ -45,6 +45,8 @@ from pypaimon.daft.daft_datasource import (
READER_MODE_NATIVE_PARQUET,
_blob_native_covering_files,
)
+from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.schema_change import SchemaChange
@dataclass
@@ -213,6 +215,39 @@ class DaftBlobNativeReadE2ETest(unittest.TestCase):
got = {r["id"]: r["name"] for r in out}
self.assertEqual(got, {i: f"n{i}" for i in range(6)})
+ def test_drop_readd_scalar_projection_falls_back(self):
+ self.catalog.alter_table(
+ self.table,
+ [SchemaChange.drop_column("name")],
+ ignore_if_not_exists=False,
+ )
+ self.catalog.alter_table(
+ self.table,
+ [SchemaChange.add_column("name", AtomicType("STRING"))],
+ ignore_if_not_exists=False,
+ )
+
+ result = explain_paimon_scan(
+ self.table,
+ self.catalog_options,
+ columns=["id", "name"],
+ verbose=True,
+ )
+ out = (
+ read_paimon(self.table, self.catalog_options)
+ .select(col("id"), col("name"))
+ .sort("id")
+ .to_pydict()
+ )
+
+ self.assertEqual(result.native_parquet_split_count, 0)
+ self.assertGreater(result.pypaimon_fallback_split_count, 0)
+ self.assertIn(
+ "schema evolution requires PyPaimon normalization",
+ result.fallback_reasons,
+ )
+ self.assertEqual(out, {"id": list(range(6)), "name": [None] * 6})
+
def test_blob_projection_falls_back(self):
result = explain_paimon_scan(
self.table, self.catalog_options, columns=["id", "content"],
verbose=True)
diff --git a/paimon-python/pypaimon/tests/daft/daft_datasource_test.py
b/paimon-python/pypaimon/tests/daft/daft_datasource_test.py
index fd49adf242..484cd052ff 100644
--- a/paimon-python/pypaimon/tests/daft/daft_datasource_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_datasource_test.py
@@ -19,12 +19,26 @@ import unittest
from dataclasses import dataclass
from typing import Optional
+import pyarrow as pa
import pytest
pypaimon = pytest.importorskip("pypaimon")
daft = pytest.importorskip("daft")
-from pypaimon.daft.daft_datasource import PaimonDataSource
+from pypaimon.daft.daft_datasource import (
+ PaimonDataSource,
+ _native_read_fields_compatible,
+ _native_read_types_compatible,
+ _promote_time32_schema_for_daft,
+)
+from pypaimon.schema.data_types import (
+ ArrayType,
+ AtomicType,
+ DataField,
+ MapType,
+ RowType,
+ VectorType,
+)
@dataclass
@@ -33,6 +47,26 @@ class _DataFile:
external_path: Optional[str] = None
+@dataclass
+class _TableSchema:
+ id: int
+ fields: list[DataField]
+
+
+class _SchemaManager:
+ def __init__(self, schemas: list[_TableSchema]):
+ self._schemas = {schema.id: schema for schema in schemas}
+
+ def get_schema(self, schema_id: int) -> _TableSchema:
+ return self._schemas[schema_id]
+
+
+@dataclass
+class _TableWithSchemas:
+ table_schema: _TableSchema
+ schema_manager: _SchemaManager
+
+
def _build_uri(warehouse_scheme: str, file_path: str) -> str:
class _Stub:
pass
@@ -96,5 +130,304 @@ class DataFilePathTest(unittest.TestCase):
)
[email protected](
+ ("file_fields", "current_fields"),
+ [
+ pytest.param(
+ [DataField(0, "id", AtomicType("INT"))],
+ [
+ DataField(0, "id", AtomicType("INT")),
+ DataField(1, "added", AtomicType("STRING")),
+ ],
+ id="add-nullable-field",
+ ),
+ pytest.param(
+ [
+ DataField(0, "id", AtomicType("INT")),
+ DataField(1, "removed", AtomicType("STRING")),
+ ],
+ [DataField(0, "id", AtomicType("INT"))],
+ id="drop-field",
+ ),
+ pytest.param(
+ [
+ DataField(0, "id", AtomicType("INT")),
+ DataField(1, "value", AtomicType("STRING")),
+ ],
+ [
+ DataField(1, "value", AtomicType("STRING")),
+ DataField(0, "id", AtomicType("INT")),
+ ],
+ id="reorder-fields",
+ ),
+ ],
+)
+def test_native_read_fields_allow_name_aligned_evolution(
+ file_fields,
+ current_fields,
+):
+ assert _native_read_fields_compatible(file_fields, current_fields)
+
+
[email protected](
+ ("file_field", "current_field"),
+ [
+ pytest.param(
+ DataField(1, "value", AtomicType("STRING")),
+ DataField(1, "renamed", AtomicType("STRING")),
+ id="rename",
+ ),
+ pytest.param(
+ DataField(1, "value", AtomicType("STRING")),
+ DataField(2, "value", AtomicType("STRING")),
+ id="drop-then-readd",
+ ),
+ ],
+)
+def test_native_read_fields_reject_field_id_mapping(file_field, current_field):
+ assert not _native_read_fields_compatible([file_field], [current_field])
+
+
[email protected](
+ ("file_type", "current_type", "expected"),
+ [
+ pytest.param(
+ AtomicType("INT", nullable=False),
+ AtomicType("INT", nullable=True),
+ True,
+ id="relax-nullability",
+ ),
+ pytest.param(
+ AtomicType("INT"),
+ AtomicType("BIGINT"),
+ True,
+ id="promote-int",
+ ),
+ pytest.param(
+ AtomicType("INT"),
+ AtomicType("INT", nullable=False),
+ False,
+ id="tighten-nullability",
+ ),
+ pytest.param(
+ AtomicType("BIGINT"),
+ AtomicType("INT"),
+ False,
+ id="narrow-int",
+ ),
+ pytest.param(
+ AtomicType("DECIMAL(10, 4)"),
+ AtomicType("DECIMAL(10, 2)"),
+ False,
+ id="decimal-scale-down",
+ ),
+ pytest.param(
+ AtomicType("TIMESTAMP(3)"),
+ AtomicType("TIME(3)"),
+ False,
+ id="timestamp-to-time",
+ ),
+ pytest.param(
+ AtomicType("TIME(3)"),
+ AtomicType("TIME(3)"),
+ False,
+ id="unsupported-time",
+ ),
+ ],
+)
+def test_native_read_atomic_type_compatibility(file_type, current_type,
expected):
+ assert _native_read_types_compatible(file_type, current_type) is expected
+
+
[email protected](
+ ("file_type", "current_type"),
+ [
+ pytest.param(
+ RowType(True, [DataField(1, "value", AtomicType("INT"))]),
+ RowType(True, [DataField(1, "value", AtomicType("BIGINT"))]),
+ id="row",
+ ),
+ pytest.param(
+ ArrayType(True, AtomicType("INT")),
+ ArrayType(True, AtomicType("BIGINT")),
+ id="array",
+ ),
+ pytest.param(
+ MapType(True, AtomicType("STRING"), AtomicType("INT")),
+ MapType(True, AtomicType("STRING"), AtomicType("BIGINT")),
+ id="map",
+ ),
+ pytest.param(
+ VectorType(True, AtomicType("INT"), 3),
+ VectorType(True, AtomicType("BIGINT"), 3),
+ id="vector",
+ ),
+ ],
+)
+def test_native_read_types_allow_recursive_promotion(file_type, current_type):
+ assert _native_read_types_compatible(file_type, current_type)
+
+
+def test_native_read_types_allow_nested_name_aligned_evolution():
+ file_fields = [
+ DataField(1, "first", AtomicType("INT")),
+ DataField(2, "second", AtomicType("STRING")),
+ ]
+ current_fields = [
+ DataField(2, "second", AtomicType("STRING")),
+ DataField(1, "first", AtomicType("INT")),
+ ]
+
+ assert _native_read_types_compatible(
+ RowType(True, file_fields),
+ RowType(True, current_fields),
+ )
+
+
+def _identity(data_type):
+ return data_type
+
+
+def _array(data_type):
+ return ArrayType(True, data_type)
+
+
+def _map_value(data_type):
+ return MapType(True, AtomicType("STRING"), data_type)
+
+
[email protected](
+ ("wrap", "file_child", "current_child"),
+ [
+ pytest.param(
+ _identity,
+ DataField(1, "value", AtomicType("INT")),
+ DataField(1, "renamed", AtomicType("INT")),
+ id="row-nested-rename",
+ ),
+ pytest.param(
+ _array,
+ DataField(1, "value", AtomicType("INT")),
+ DataField(2, "value", AtomicType("INT")),
+ id="array-row-nested-drop-then-readd",
+ ),
+ pytest.param(
+ _map_value,
+ DataField(1, "value", AtomicType("INT")),
+ DataField(1, "renamed", AtomicType("INT")),
+ id="map-row-nested-rename",
+ ),
+ ],
+)
+def test_native_read_types_reject_nested_field_id_mapping(
+ wrap,
+ file_child,
+ current_child,
+):
+ file_type = wrap(RowType(True, [file_child]))
+ current_type = wrap(RowType(True, [current_child]))
+
+ assert not _native_read_types_compatible(file_type, current_type)
+
+
+def test_native_read_types_reject_vector_length_change():
+ assert not _native_read_types_compatible(
+ VectorType(True, AtomicType("INT"), 3),
+ VectorType(True, AtomicType("INT"), 4),
+ )
+
+
+def _row_value(data_type):
+ return RowType(True, [DataField(1, "value", data_type)])
+
+
[email protected](
+ ("wrap", "file_type", "current_type"),
+ [
+ pytest.param(
+ _row_value,
+ AtomicType("DECIMAL(10, 4)"),
+ AtomicType("DECIMAL(10, 2)"),
+ id="row-decimal-scale-down",
+ ),
+ pytest.param(
+ _array,
+ AtomicType("TIMESTAMP(3)"),
+ AtomicType("TIME(3)"),
+ id="array-timestamp-to-time",
+ ),
+ pytest.param(
+ _map_value,
+ AtomicType("DECIMAL(10, 4)"),
+ AtomicType("DECIMAL(10, 2)"),
+ id="map-decimal-scale-down",
+ ),
+ ],
+)
+def test_native_read_types_reject_semantic_mismatch_recursively(
+ wrap,
+ file_type,
+ current_type,
+):
+ assert not _native_read_types_compatible(
+ wrap(file_type),
+ wrap(current_type),
+ )
+
+
+def test_promote_time32_schema_for_daft_recursively():
+ schema = pa.schema(
+ [
+ ("value", pa.time32("ms")),
+ ("row", pa.struct([("value", pa.time32("ms"))])),
+ ("array", pa.list_(pa.time32("ms"))),
+ ("map", pa.map_(pa.string(), pa.time32("ms"))),
+ ]
+ )
+
+ assert _promote_time32_schema_for_daft(schema) == pa.schema(
+ [
+ ("value", pa.time64("us")),
+ ("row", pa.struct([("value", pa.time64("us"))])),
+ ("array", pa.list_(pa.time64("us"))),
+ ("map", pa.map_(pa.string(), pa.time64("us"))),
+ ]
+ )
+
+
+def test_schema_compatibility_uses_task_columns_for_mixed_schema_split():
+ old_schema = _TableSchema(
+ 0,
+ [
+ DataField(0, "id", AtomicType("BIGINT")),
+ DataField(1, "value", AtomicType("STRING")),
+ ],
+ )
+ current_schema = _TableSchema(
+ 1,
+ [
+ DataField(0, "id", AtomicType("BIGINT")),
+ DataField(1, "renamed", AtomicType("STRING")),
+ ],
+ )
+ table = _TableWithSchemas(
+ current_schema,
+ _SchemaManager([old_schema, current_schema]),
+ )
+
+ assert not PaimonDataSource._has_incompatible_file_schema(
+ table,
+ [current_schema.id, old_schema.id],
+ ["id"],
+ {},
+ )
+ assert PaimonDataSource._has_incompatible_file_schema(
+ table,
+ [current_schema.id, old_schema.id],
+ ["renamed"],
+ {},
+ )
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/paimon-python/pypaimon/tests/daft/daft_explain_test.py
b/paimon-python/pypaimon/tests/daft/daft_explain_test.py
index 6f3a54a62b..8ed2bfaf3e 100644
--- a/paimon-python/pypaimon/tests/daft/daft_explain_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_explain_test.py
@@ -38,6 +38,7 @@ from pypaimon.daft.daft_compat import has_file_range_reads
from pypaimon.daft.daft_datasource import PaimonDataSource
from pypaimon.daft.daft_paimon import _explain_table
from pypaimon.read.explain import ExplainResult, ExplainSplitInfo
+from pypaimon.schema.schema_change import SchemaChange
requires_blob = pytest.mark.skipif(not has_file_range_reads(), reason="BLOB
support requires daft >= 0.7.11")
@@ -172,6 +173,115 @@ def
test_explain_paimon_scan_reports_native_parquet_routing(catalog_options):
assert "PyPaimon Scan Plan" in str(result)
+def test_explain_scan_reports_schema_evolution_fallback(catalog_options):
+ old_schema = pa.schema([
+ ("id", pa.int64()),
+ ("value", pa.string()),
+ ])
+ identifier, table = _create_table(
+ catalog_options,
+ "explain_schema_evolution_fallback",
+ old_schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ old_schema_id = table.table_schema.id
+ _write_arrow(
+ table,
+ pa.table({"id": [1], "value": ["a"]}, schema=old_schema),
+ )
+
+ catalog = pypaimon.CatalogFactory.create(catalog_options)
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.rename_column("value", "renamed")],
+ ignore_if_not_exists=False,
+ )
+
+ projected_result = explain_paimon_scan(
+ identifier,
+ catalog_options,
+ columns=["id"],
+ verbose=True,
+ )
+ filtered_result = explain_paimon_scan(
+ identifier,
+ catalog_options,
+ filters=col("renamed") == "a",
+ columns=["id"],
+ verbose=True,
+ )
+ assert projected_result.native_parquet_split_count == 1
+ assert projected_result.pypaimon_fallback_split_count == 0
+ assert projected_result.task_columns == ["id"]
+ assert filtered_result.pypaimon_fallback_split_count == 1
+ assert filtered_result.native_parquet_split_count == 0
+ assert filtered_result.task_columns is not None
+ assert set(filtered_result.task_columns) == {"id", "renamed"}
+ assert filtered_result.fallback_reasons == {
+ "schema evolution requires PyPaimon normalization": 1,
+ }
+ assert filtered_result.paimon_scan.splits is not None
+ assert filtered_result.paimon_scan.splits[0].data_files is not None
+ assert {
+ data_file.schema_id
+ for data_file in filtered_result.paimon_scan.splits[0].data_files
+ } == {old_schema_id}
+ assert filtered_result.splits is not None
+ assert filtered_result.splits[0].reader_mode ==
READER_MODE_PYPAIMON_FALLBACK
+ assert filtered_result.splits[0].fallback_reason == (
+ "schema evolution requires PyPaimon normalization"
+ )
+
+
+def test_explain_scan_keeps_native_route_after_column_comment_change(
+ catalog_options,
+):
+ pa_schema = pa.schema([
+ ("id", pa.int64()),
+ ("value", pa.string()),
+ ])
+ identifier, table = _create_table(
+ catalog_options,
+ "explain_column_comment_change",
+ pa_schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ old_schema_id = table.table_schema.id
+ _write_arrow(
+ table,
+ pa.table({"id": [1], "value": ["a"]}, schema=pa_schema),
+ )
+
+ catalog = pypaimon.CatalogFactory.create(catalog_options)
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.update_column_comment("value", "updated comment")],
+ ignore_if_not_exists=False,
+ )
+ table = catalog.get_table(identifier)
+
+ result = explain_paimon_scan(
+ identifier,
+ catalog_options,
+ verbose=True,
+ )
+
+ assert table.table_schema.id != old_schema_id
+ assert table.fields[1].description == "updated comment"
+ assert result.native_parquet_split_count == 1
+ assert result.pypaimon_fallback_split_count == 0
+ assert result.fallback_reasons == {}
+ assert result.paimon_scan.splits is not None
+ assert result.paimon_scan.splits[0].data_files is not None
+ assert {
+ data_file.schema_id
+ for data_file in result.paimon_scan.splits[0].data_files
+ } == {old_schema_id}
+ assert result.splits is not None
+ assert result.splits[0].reader_mode == READER_MODE_NATIVE_PARQUET
+ assert result.splits[0].fallback_reason is None
+
+
def test_explain_scan_keeps_limit_above_remaining_filters(catalog_options):
pa_schema = pa.schema([
("id", pa.int64()),
diff --git a/paimon-python/pypaimon/tests/daft/daft_integration_test.py
b/paimon-python/pypaimon/tests/daft/daft_integration_test.py
index 0a8e44466c..de31cdfc49 100644
--- a/paimon-python/pypaimon/tests/daft/daft_integration_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_integration_test.py
@@ -21,6 +21,7 @@
from __future__ import annotations
import datetime
+import decimal
import time
import pyarrow as pa
@@ -31,8 +32,10 @@ daft = pytest.importorskip("daft")
from daft import col
-from pypaimon.daft import read_paimon, write_paimon
+from pypaimon.daft import explain_paimon_scan, read_paimon, write_paimon
from pypaimon.daft.daft_paimon import _timestamp_scan_option
+from pypaimon.schema.data_types import AtomicType
+from pypaimon.schema.schema_change import Move, SchemaChange
@pytest.fixture
@@ -164,6 +167,369 @@ def
test_read_paimon_data_evolution_merges_column_fragments(catalog_options):
}
+def test_read_paimon_renamed_column_from_old_schema_file(catalog_options):
+ old_schema = pa.schema([
+ ("id", pa.int64()),
+ ("value", pa.string()),
+ ])
+ identifier, table = _create_table(
+ catalog_options,
+ "read_renamed_column",
+ old_schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ old_schema_id = table.table_schema.id
+ _write_arrow(
+ table,
+ pa.table({"id": [1, 2], "value": ["a", "b"]}, schema=old_schema),
+ )
+
+ catalog = pypaimon.CatalogFactory.create(catalog_options)
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.rename_column("value", "renamed")],
+ ignore_if_not_exists=False,
+ )
+ table = catalog.get_table(identifier)
+ splits = table.new_read_builder().new_scan().plan().splits()
+
+ assert all(split.raw_convertible for split in splits)
+ assert table.table_schema.id != old_schema_id
+ assert {
+ data_file.schema_id
+ for split in splits
+ for data_file in split.files
+ } == {old_schema_id}
+
+ projected_result = (
+ read_paimon(identifier, catalog_options)
+ .select("id")
+ .sort("id")
+ .to_pydict()
+ )
+ result = read_paimon(identifier, catalog_options).sort("id").to_pydict()
+
+ assert projected_result == {"id": [1, 2]}
+ assert result == {
+ "id": [1, 2],
+ "renamed": ["a", "b"],
+ }
+
+
+def
test_read_paimon_renamed_nested_field_from_old_schema_file(catalog_options):
+ old_schema = pa.schema([
+ ("id", pa.int64()),
+ ("payload", pa.struct([
+ ("value", pa.string()),
+ ("count", pa.int32()),
+ ])),
+ ])
+ identifier, table = _create_table(
+ catalog_options,
+ "read_renamed_nested_field",
+ old_schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ _write_arrow(
+ table,
+ pa.Table.from_pylist(
+ [
+ {"id": 1, "payload": {"value": "a", "count": 10}},
+ {"id": 2, "payload": {"value": "b", "count": 20}},
+ ],
+ schema=old_schema,
+ ),
+ )
+
+ catalog = pypaimon.CatalogFactory.create(catalog_options)
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.rename_column(["payload", "value"], "renamed")],
+ ignore_if_not_exists=False,
+ )
+
+ explain = explain_paimon_scan(identifier, catalog_options, verbose=True)
+ result = read_paimon(identifier, catalog_options).sort("id").to_pydict()
+
+ assert explain.pypaimon_fallback_split_count == 1
+ assert explain.native_parquet_split_count == 0
+ assert result == {
+ "id": [1, 2],
+ "payload": [
+ {"renamed": "a", "count": 10},
+ {"renamed": "b", "count": 20},
+ ],
+ }
+
+
+def test_read_paimon_drop_readd_same_column_isolates_field_ids(
+ catalog_options,
+):
+ old_schema = pa.schema([
+ ("id", pa.int64()),
+ ("value", pa.string()),
+ ])
+ identifier, table = _create_table(
+ catalog_options,
+ "read_drop_readd_column",
+ old_schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ old_value_field = table.fields[1]
+ _write_arrow(
+ table,
+ pa.table({"id": [1, 2], "value": ["a", "b"]}, schema=old_schema),
+ )
+
+ catalog = pypaimon.CatalogFactory.create(catalog_options)
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.drop_column("value")],
+ ignore_if_not_exists=False,
+ )
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.add_column("value", AtomicType("STRING"))],
+ ignore_if_not_exists=False,
+ )
+ table = catalog.get_table(identifier)
+ new_value_field = table.fields[1]
+
+ assert new_value_field.name == old_value_field.name
+ assert new_value_field.type == old_value_field.type
+ assert new_value_field.id != old_value_field.id
+
+ explain = explain_paimon_scan(identifier, catalog_options, verbose=True)
+
+ assert explain.pypaimon_fallback_split_count == 1
+ assert explain.native_parquet_split_count == 0
+
+ result = read_paimon(identifier, catalog_options).sort("id").to_pydict()
+
+ assert result == {
+ "id": [1, 2],
+ "value": [None, None],
+ }
+
+
+def test_native_read_handles_name_aligned_schema_evolution(catalog_options):
+ old_schema = pa.schema([
+ pa.field("id", pa.int32(), nullable=False),
+ pa.field("value", pa.string()),
+ ])
+ identifier, table = _create_table(
+ catalog_options,
+ "read_native_schema_evolution",
+ old_schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ _write_arrow(
+ table,
+ pa.table({"id": [1, 2], "value": ["a", "b"]}, schema=old_schema),
+ )
+
+ catalog = pypaimon.CatalogFactory.create(catalog_options)
+ catalog.alter_table(
+ identifier,
+ [
+ SchemaChange.add_column("added", AtomicType("STRING")),
+ SchemaChange.update_column_position(Move.first("added")),
+ SchemaChange.update_column_nullability("id", True),
+ SchemaChange.update_column_type("id", AtomicType("BIGINT")),
+ SchemaChange.drop_column("value"),
+ ],
+ ignore_if_not_exists=False,
+ )
+
+ explain = explain_paimon_scan(identifier, catalog_options, verbose=True)
+ result = read_paimon(identifier, catalog_options).sort("id").to_pydict()
+ filtered_result = (
+ read_paimon(identifier, catalog_options)
+ .where(col("id") == 2)
+ .select("id")
+ .to_pydict()
+ )
+
+ assert explain.native_parquet_split_count == 1
+ assert explain.pypaimon_fallback_split_count == 0
+ assert explain.fallback_reasons == {}
+ assert result == {
+ "added": [None, None],
+ "id": [1, 2],
+ }
+ assert filtered_result == {"id": [2]}
+
+
+def test_native_read_handles_nested_type_widening(catalog_options):
+ old_schema = pa.schema([
+ ("id", pa.int64()),
+ ("payload", pa.struct([
+ ("value", pa.int32()),
+ ("label", pa.string()),
+ ])),
+ ])
+ identifier, table = _create_table(
+ catalog_options,
+ "read_nested_type_widening",
+ old_schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ _write_arrow(
+ table,
+ pa.Table.from_pylist(
+ [
+ {"id": 1, "payload": {"value": 10, "label": "a"}},
+ {"id": 2, "payload": {"value": 20, "label": "b"}},
+ ],
+ schema=old_schema,
+ ),
+ )
+
+ catalog = pypaimon.CatalogFactory.create(catalog_options)
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.update_column_type(["payload", "value"],
AtomicType("BIGINT"))],
+ ignore_if_not_exists=False,
+ )
+
+ explain = explain_paimon_scan(identifier, catalog_options, verbose=True)
+ result = read_paimon(identifier, catalog_options).sort("id").to_arrow()
+
+ assert explain.native_parquet_split_count == 1
+ assert explain.pypaimon_fallback_split_count == 0
+ assert explain.fallback_reasons == {}
+ assert result.schema.field("payload").type.field("value").type ==
pa.int64()
+ assert result.to_pydict() == {
+ "id": [1, 2],
+ "payload": [
+ {"value": 10, "label": "a"},
+ {"value": 20, "label": "b"},
+ ],
+ }
+
+
[email protected](
+ (
+ "case_name",
+ "old_type",
+ "pypaimon_type",
+ "daft_type",
+ "new_paimon_type",
+ "values",
+ "expected",
+ ),
+ [
+ pytest.param(
+ "decimal_scale_down",
+ pa.decimal128(10, 4),
+ pa.decimal128(10, 2),
+ pa.decimal128(10, 2),
+ AtomicType("DECIMAL(10, 2)"),
+ [decimal.Decimal("1.2355"), decimal.Decimal("-4.5678")],
+ [decimal.Decimal("1.23"), decimal.Decimal("-4.56")],
+ id="decimal-scale-down",
+ ),
+ pytest.param(
+ "timestamp_to_time",
+ pa.timestamp("ms"),
+ pa.time32("ms"),
+ pa.time64("us"),
+ AtomicType("TIME(3)"),
+ [
+ datetime.datetime(2025, 1, 2, 3, 4, 5, 678000),
+ datetime.datetime(1999, 12, 31, 23, 59, 58, 987000),
+ ],
+ [
+ datetime.time(3, 4, 5, 678000),
+ datetime.time(23, 59, 58, 987000),
+ ],
+ id="timestamp-to-time",
+ ),
+ ],
+)
+def test_read_paimon_falls_back_for_native_cast_semantic_mismatch(
+ catalog_options,
+ case_name,
+ old_type,
+ pypaimon_type,
+ daft_type,
+ new_paimon_type,
+ values,
+ expected,
+):
+ old_schema = pa.schema([("id", pa.int64()), ("value", old_type)])
+ identifier, table = _create_table(
+ catalog_options,
+ f"read_{case_name}",
+ old_schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ _write_arrow(
+ table,
+ pa.table({"id": [1, 2], "value": values}, schema=old_schema),
+ )
+
+ catalog = pypaimon.CatalogFactory.create(catalog_options)
+ catalog.alter_table(
+ identifier,
+ [SchemaChange.update_column_type("value", new_paimon_type)],
+ ignore_if_not_exists=False,
+ )
+ table = catalog.get_table(identifier)
+
+ explain = explain_paimon_scan(identifier, catalog_options, verbose=True)
+ daft_result = read_paimon(identifier,
catalog_options).sort("id").to_arrow()
+ read_builder = table.new_read_builder()
+ pypaimon_result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits()
+ ).sort_by([("id", "ascending")])
+ expected_result = pa.table(
+ {"id": [1, 2], "value": expected},
+ schema=pa.schema([("id", pa.int64()), ("value", pypaimon_type)]),
+ )
+
+ assert explain.pypaimon_fallback_split_count == 1
+ assert explain.native_parquet_split_count == 0
+ assert daft_result.schema == pa.schema(
+ [("id", pa.int64()), ("value", daft_type)]
+ )
+ assert daft_result.to_pydict() == pypaimon_result.to_pydict()
+ assert pypaimon_result.equals(expected_result)
+
+
+def test_read_paimon_time_uses_pypaimon_fallback(catalog_options):
+ schema = pa.schema([("id", pa.int64()), ("value", pa.time32("ms"))])
+ values = [
+ datetime.time(3, 4, 5, 678000),
+ datetime.time(23, 59, 58, 987000),
+ ]
+ identifier, table = _create_table(
+ catalog_options,
+ "read_time",
+ schema,
+ options={"bucket": "-1", "file.format": "parquet"},
+ )
+ _write_arrow(
+ table,
+ pa.table({"id": [1, 2], "value": values}, schema=schema),
+ )
+
+ explain = explain_paimon_scan(identifier, catalog_options, verbose=True)
+ daft_result = read_paimon(identifier,
catalog_options).sort("id").to_arrow()
+ read_builder = table.new_read_builder()
+ pypaimon_result = read_builder.new_read().to_arrow(
+ read_builder.new_scan().plan().splits()
+ ).sort_by([("id", "ascending")])
+
+ assert explain.pypaimon_fallback_split_count == 1
+ assert explain.native_parquet_split_count == 0
+ assert daft_result.schema == pa.schema(
+ [("id", pa.int64()), ("value", pa.time64("us"))]
+ )
+ assert pypaimon_result.schema == schema
+ assert daft_result.to_pydict() == pypaimon_result.to_pydict()
+
+
def test_read_paimon_projection(catalog_options):
data = pa.table(
{
