From ed00358d52ef1bd4c1022b841d35dc9a3772fb2d Mon Sep 17 00:00:00 2001 From: Yann Date: Thu, 15 Oct 2026 11:50:12 +0800 Subject: [PATCH 1/4] fix(python): align empty and metadata-only results with runtime read schemas Use the core read schema for Ray empty results and metadata-only search results. Keep read expectations separate from logical write constraints and preserve strict value, schema, and invalid-NULL checks. Co-Authored-By: Codex AI-Model: gpt-6 Co-Authored-By: Codex AI-Contributed/Feature: 14/14 AI-Contributed/UT: 202/202 --- paimon-python/pypaimon/multimodal/query.py | 3 +- .../pypaimon/ray/data_evolution_merge_join.py | 11 ++------ .../pypaimon/tests/binary_row_test.py | 25 +++++++++-------- .../pypaimon/tests/blob_table_test.py | 13 +++++---- .../pypaimon/tests/data_evolution_test.py | 19 ++++++------- .../pypaimon/tests/filesystem_catalog_test.py | 28 ++++--------------- .../tests/map_shared_shredding_write_test.py | 7 ++++- .../tests/native_cast_completion_test.py | 8 ++++-- .../tests/native_system_columns_test.py | 7 ++++- .../pypaimon/tests/ray_read_by_row_id_test.py | 22 ++++++++++++++- .../pypaimon/tests/reader_primary_key_test.py | 11 ++++---- .../tests/rest/rest_read_write_test.py | 15 +++++----- .../pypaimon/tests/rest/rest_simple_test.py | 5 ++-- .../tests/search_result_metadata_test.py | 17 +++++++++++ .../pypaimon/tests/serializable_test.py | 3 +- .../pypaimon/tests/table_commit_test.py | 4 ++- .../tests/table_upsert_by_key_test.py | 3 +- .../pypaimon/tests/write/table_write_test.py | 15 ++++++---- 18 files changed, 130 insertions(+), 86 deletions(-) diff --git a/paimon-python/pypaimon/multimodal/query.py b/paimon-python/pypaimon/multimodal/query.py index 13aa615b8054..bba7073da945 100644 --- a/paimon-python/pypaimon/multimodal/query.py +++ b/paimon-python/pypaimon/multimodal/query.py @@ -20,6 +20,7 @@ import pyarrow as pa +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.common.where_parser import parse_where_clause from pypaimon.multimodal.blob_read import fetch_blob_bodies from pypaimon.schema.data_types import is_array_blob_type, is_blob_type, is_map_blob_type @@ -533,7 +534,7 @@ def _read_global_index_result(self, result): added_row_id = projection is None or row_id_name not in projection if metadata_only: row_ids = pa.array(list(result.results()), type=pa.int64()) - fields = [pa.field(row_id_name, pa.int64(), nullable=False)] * (len(projection) or 1) + fields = [read_arrow_schema([SpecialFields.ROW_ID]).field(row_id_name)] * (len(projection) or 1) table = pa.Table.from_arrays( [row_ids] * len(fields), schema=pa.schema(fields)) else: diff --git a/paimon-python/pypaimon/ray/data_evolution_merge_join.py b/paimon-python/pypaimon/ray/data_evolution_merge_join.py index c745db9e8e6c..92ce70995410 100644 --- a/paimon-python/pypaimon/ray/data_evolution_merge_join.py +++ b/paimon-python/pypaimon/ray/data_evolution_merge_join.py @@ -877,16 +877,11 @@ def _apply_group(group: pa.Table) -> pa.Table: def _read_output_schema(table, read_cols: Sequence[str]) -> "pa.Schema": """Result schema: each projected column's type plus int64 ``_ROW_ID``, in ``read_cols`` order. Shared by the empty-result paths so they can't drift.""" - from pypaimon.schema.data_types import PyarrowFieldParser + from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.table.special_fields import SpecialFields - rid = SpecialFields.ROW_ID.name - full = PyarrowFieldParser.from_paimon_schema(table.table_schema.fields) - # Keep each field's nullability so an empty result matches a non-empty read. - return pa.schema([ - pa.field(rid, pa.int64(), nullable=False) if col == rid else full.field(col) - for col in read_cols - ]) + full = read_arrow_schema(table.table_schema.fields + [SpecialFields.ROW_ID]) + return pa.schema([full.field(col) for col in read_cols]) def distributed_read_by_row_id( diff --git a/paimon-python/pypaimon/tests/binary_row_test.py b/paimon-python/pypaimon/tests/binary_row_test.py index 5e8420e9eb42..5d4efa18095c 100644 --- a/paimon-python/pypaimon/tests/binary_row_test.py +++ b/paimon-python/pypaimon/tests/binary_row_test.py @@ -26,6 +26,7 @@ import pytest from pypaimon import CatalogFactory, Schema +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.manifest.schema.manifest_entry import ManifestEntry from pypaimon.manifest.schema.simple_stats import SimpleStats from pypaimon.read.scanner.file_scanner import FileScanner @@ -78,6 +79,8 @@ def setUpClass(cls): write.close() commit.close() + cls.expected = cls.data.cast(read_arrow_schema(append_table.fields)) + @classmethod def tearDownClass(cls): shutil.rmtree(cls.tempdir, ignore_errors=True) @@ -90,7 +93,7 @@ def test_not_equal_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.not_equal('f2', 6) # test stats filter when filtering ManifestEntry splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.slice(1, 4) + expected = self.expected.slice(1, 4) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) @@ -104,7 +107,7 @@ def test_less_than_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.less_than('f2', 8) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.slice(0, 2) + expected = self.expected.slice(0, 2) self.assertEqual(actual, expected) self.assertEqual(len(expected), len(splits)) # test stats filter when filtering ManifestEntry @@ -114,7 +117,7 @@ def test_is_null_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_null('f1') # value_stats_cols=None splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.slice(4, 1) + expected = self.expected.slice(4, 1) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) @@ -140,7 +143,7 @@ def test_is_not_null_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_not_null('f1') splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.slice(0, 4) + expected = self.expected.slice(0, 4) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) @@ -152,7 +155,7 @@ def test_is_in_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_in('f2', [6, 8]) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.take([0, 2]) + expected = self.expected.take([0, 2]) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) @@ -162,7 +165,7 @@ def test_equal_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.equal('f2', 6) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.slice(0, 1) + expected = self.expected.slice(0, 1) self.assertEqual(expected, actual) self.assertEqual(len(splits), len(expected)) # test partition filter when filtering ManifestEntry @@ -173,7 +176,7 @@ def test_not_equal_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.not_equal('f2', 6) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.slice(1, 4) + expected = self.expected.slice(1, 4) self.assertEqual(actual, expected) self.assertEqual(len(splits), len(expected)) # test partition filter when filtering ManifestEntry @@ -184,7 +187,7 @@ def test_less_than_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.less_than('f0', 3) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.slice(0, 2) + expected = self.expected.slice(0, 2) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) # test key stats filter when filtering ManifestEntry @@ -194,7 +197,7 @@ def test_is_null_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_null('f1') splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.slice(4, 1) + expected = self.expected.slice(4, 1) self.assertEqual(actual, expected) @pytest.mark.python_plan @@ -204,7 +207,7 @@ def test_is_not_null_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_not_null('f1') splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.data.slice(0, 4) + expected = self.expected.slice(0, 4) self.assertEqual(actual, expected) @pytest.mark.python_plan @@ -215,7 +218,7 @@ def test_is_in_pk(self): predicate = predicate_builder.is_in('f0', [1, 5]) splits, actual = self._read_result(read_builder.with_filter(predicate)) # expected rows: indices [0, 3] - expected = self.data.take([0, 4]) + expected = self.expected.take([0, 4]) self.assertEqual(actual, expected) self.assertEqual(len(splits), len(expected)) # test key stats filter when filtering ManifestEntry diff --git a/paimon-python/pypaimon/tests/blob_table_test.py b/paimon-python/pypaimon/tests/blob_table_test.py index 58e2484a60f7..8818e85a071f 100755 --- a/paimon-python/pypaimon/tests/blob_table_test.py +++ b/paimon-python/pypaimon/tests/blob_table_test.py @@ -26,6 +26,7 @@ import pytest from pypaimon import CatalogFactory, Schema +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.schema.schema_change import SchemaChange from pypaimon.table.file_store_table import FileStoreTable from pypaimon.table.row.blob import Blob @@ -1716,7 +1717,8 @@ def test_map_blob_value_not_null(self): read_builder = table.new_read_builder() result = read_builder.new_read().to_arrow( read_builder.new_scan().plan().splits()) - self.assertFalse(result.schema.field('payloads').type.item_field.nullable) + self.assertFalse(table.field_dict['payloads'].type.value.nullable) + self.assertEqual(result.schema, read_arrow_schema(table.fields)) self.assertEqual( { row['id']: None if row['payloads'] is None else dict(row['payloads']) @@ -1739,7 +1741,7 @@ def test_map_blob_value_not_null(self): 'payloads': pa.array([[('bad', None)]], type=nullable_map_type), }, schema=incompatible_schema) incompatible_writer = table.new_batch_write_builder().new_write() - with self.assertRaisesRegex(ValueError, "Input schema isn't consistent"): + with self.assertRaisesRegex(ValueError, "Input schema isn't consistent|non-nullable"): incompatible_writer.write_arrow(incompatible_data) incompatible_writer.abort() @@ -1787,9 +1789,8 @@ def test_array_blob_element_not_null(self): read_builder = table.new_read_builder() result = read_builder.new_read().to_arrow( read_builder.new_scan().plan().splits()) - self.assertFalse( - result.schema.field('payloads').type.value_field.nullable - ) + self.assertFalse(table.field_dict['payloads'].type.element.nullable) + self.assertEqual(result.schema, read_arrow_schema(table.fields)) self.assertEqual( { row['id']: row['payloads'] @@ -1812,7 +1813,7 @@ def test_array_blob_element_not_null(self): 'payloads': pa.array([[b'bad', None]], type=nullable_array_type), }, schema=incompatible_schema) incompatible_writer = table.new_batch_write_builder().new_write() - with self.assertRaisesRegex(ValueError, "Input schema isn't consistent"): + with self.assertRaisesRegex(ValueError, "Input schema isn't consistent|non-nullable"): incompatible_writer.write_arrow(incompatible_data) incompatible_writer.abort() diff --git a/paimon-python/pypaimon/tests/data_evolution_test.py b/paimon-python/pypaimon/tests/data_evolution_test.py index 99af2e4933b0..38d7c131c31a 100644 --- a/paimon-python/pypaimon/tests/data_evolution_test.py +++ b/paimon-python/pypaimon/tests/data_evolution_test.py @@ -28,9 +28,11 @@ import pytest from pypaimon import CatalogFactory, Schema +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.common.predicate import Predicate from pypaimon.manifest.manifest_list_manager import ManifestListManager from pypaimon.table.row.offset_row import OffsetRow +from pypaimon.table.special_fields import SpecialFields def _filter_batch_arrow(batch, predicate): @@ -1049,14 +1051,11 @@ def test_multiple_appends_different_first_row_ids(self): table_scan = read_builder.new_scan() table_read = read_builder.new_read() actual_with_meta = table_read.to_arrow(table_scan.plan().splits()) - self.assertFalse( - actual_with_meta.schema.field('_ROW_ID').nullable, - '_ROW_ID must be non-nullable per SpecialFields', - ) - self.assertFalse( - actual_with_meta.schema.field('_SEQUENCE_NUMBER').nullable, - '_SEQUENCE_NUMBER must be non-nullable per SpecialFields', - ) + self.assertFalse(SpecialFields.ROW_ID.type.nullable) + self.assertFalse(SpecialFields.SEQUENCE_NUMBER.type.nullable) + self.assertEqual(actual_with_meta.schema, read_arrow_schema(read_builder.read_type())) + self.assertEqual(actual_with_meta.column('_ROW_ID').null_count, 0) + self.assertEqual(actual_with_meta.column('_SEQUENCE_NUMBER').null_count, 0) rb_with_row_id = table.new_read_builder().with_projection(['f0', 'f1', 'f2', '_ROW_ID']) pb = rb_with_row_id.new_predicate_builder() @@ -1246,6 +1245,7 @@ def test_read_row_tracking_metadata(self): ('f1', pa.int16()), pa.field('_SEQUENCE_NUMBER', pa.int64(), nullable=False), ])) + expect_data = expect_data.cast(read_arrow_schema(read_builder.read_type())) self.assertEqual(actual_data, expect_data) self.assertEqual(len(actual_data.schema), len(expect_data.schema), 'Read output column count must match schema') @@ -1269,8 +1269,6 @@ def test_read_row_tracking_metadata(self): table_scan = read_builder.new_scan() table_read = read_builder.new_read() actual_data = table_read.to_arrow(table_scan.plan().splits()) - self.assertFalse(actual_data.schema.field('_ROW_ID').nullable) - self.assertFalse(actual_data.schema.field('_SEQUENCE_NUMBER').nullable) expect_data = pa.Table.from_pydict({ 'f0': [3, 4], 'f1': [-1001, 1002], @@ -1282,6 +1280,7 @@ def test_read_row_tracking_metadata(self): pa.field('_ROW_ID', pa.int64(), nullable=False), pa.field('_SEQUENCE_NUMBER', pa.int64(), nullable=False), ])) + expect_data = expect_data.cast(read_arrow_schema(read_builder.read_type())) self.assertEqual(actual_data, expect_data) self.assertEqual(len(actual_data.schema), len(expect_data.schema), 'Read output column count must match schema') diff --git a/paimon-python/pypaimon/tests/filesystem_catalog_test.py b/paimon-python/pypaimon/tests/filesystem_catalog_test.py index 45ced30f6ed0..4690ffeefe9e 100644 --- a/paimon-python/pypaimon/tests/filesystem_catalog_test.py +++ b/paimon-python/pypaimon/tests/filesystem_catalog_test.py @@ -417,40 +417,24 @@ def test_update_column_type_rejects_non_executable_cast(self): catalog = CatalogFactory.create({"warehouse": self.warehouse}) catalog.create_database("test_db_cast", False) - identifier = "test_db_cast.ts_table" + identifier = "test_db_cast.time_table" schema = Schema( fields=[ DataField.from_dict({"id": 0, "name": "k", "type": "INT"}), - DataField.from_dict({"id": 1, "name": "ts", "type": "TIMESTAMP(3)"}), + DataField.from_dict({"id": 1, "name": "t", "type": "TIME(3)"}), ], partition_keys=[], primary_keys=[], options={}, comment="", ) catalog.create_table(identifier, schema, False) - # TIMESTAMP -> DECIMAL is logically allowed but has no PyArrow cast - # kernel, so the read path could not materialize it. Reject at alter - # time (mirrors Java's CastExecutors.resolve(...) != null check) instead - # of failing later at read with ArrowNotImplementedError. - with self.assertRaises(RuntimeError) as ctx: - catalog.alter_table( - identifier, - [SchemaChange.update_column_type( - "ts", AtomicType("DECIMAL(10, 0)"))], - False) - self.assertIn("no executable cast", str(ctx.exception)) - - # INT -> DECIMAL(10, 2) has a PyArrow kernel but the target precision is - # too small to hold an int's range at scale 2 (needs >= 12); the read - # path would fail with ArrowInvalid, so reject it at alter time too. - with self.assertRaises(RuntimeError) as ctx: + # TIME precision changes have no executor in either reader backend. + with self.assertRaisesRegex(RuntimeError, "no executable cast"): catalog.alter_table( identifier, - [SchemaChange.update_column_type( - "k", AtomicType("DECIMAL(10, 2)"))], + [SchemaChange.update_column_type("t", AtomicType("TIME(1)"))], False) - self.assertIn("no executable cast", str(ctx.exception)) - # A wide-enough DECIMAL is executable and succeeds. + # INT -> DECIMAL(12, 2) is executable in both reader backends. catalog.alter_table( identifier, [SchemaChange.update_column_type("k", AtomicType("DECIMAL(12, 2)"))], diff --git a/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py b/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py index 2cbc44cace4a..13d13b9ff50b 100644 --- a/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py +++ b/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py @@ -26,6 +26,7 @@ import pyarrow.parquet as pq from pypaimon import CatalogFactory, Schema +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.data.map_shared_shredding import ( is_shared_shredding, parse_shared_shredding_metadata, @@ -169,8 +170,12 @@ def test_non_nullable_values(self): self.assertTrue(field.type['__col_1'].nullable) reader = table.new_read_builder() result = reader.new_read().to_arrow(reader.new_scan().plan().splits()) + self.assertFalse(table.field_dict['metrics'].type.value.nullable) + expected_schema = read_arrow_schema(table.fields) + self.assertEqual(result.schema, expected_schema) # Arrow 6 cannot convert non-nullable MAP values to scalars. - actual = result.column('metrics').cast(pa.map_(pa.string(), value_type)) + read_value_type = expected_schema.field('metrics').type.item_type + actual = result.column('metrics').cast(pa.map_(pa.string(), read_value_type)) self.assertEqual(expected, actual.to_pylist()) def test_nested_non_nullable_map_values_in_subprocess(self): diff --git a/paimon-python/pypaimon/tests/native_cast_completion_test.py b/paimon-python/pypaimon/tests/native_cast_completion_test.py index f047c7993777..e7b05fd78dcd 100644 --- a/paimon-python/pypaimon/tests/native_cast_completion_test.py +++ b/paimon-python/pypaimon/tests/native_cast_completion_test.py @@ -32,6 +32,7 @@ from pypaimon import Schema from pypaimon.api.api_response import AuthTableQueryResponse +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.common.predicate_builder import PredicateBuilder from pypaimon.schema.data_types import DataTypeParser, PyarrowFieldParser from pypaimon.schema.schema_change import SchemaChange @@ -193,12 +194,15 @@ def test_not_null_mask_adapts_only_output_schema(rest_catalog, table_mode, mask) expected = [None, None] if mask == 'null' else [Decimal('1.00'), None] result = _read(table, ['masked', 'id']).sort_by([('id', 'ascending')]) assert result.column('masked').to_pylist() == expected - assert result.schema.field('masked').nullable and not result.schema.field('id').nullable + assert not table.field_dict['masked'].type.nullable + assert not table.field_dict['id'].type.nullable + id_field = read_arrow_schema(table.fields).field('id') + assert result.schema == pa.schema([pa.field('masked', target), id_field]) assert _read(table, ['masked'], limit=0).schema.field('masked').nullable projection = {'first': 'masked', 'second': 'masked', 'id': 'id'} result = _read(table, projection).sort_by([('id', 'ascending')]) assert result.column('first').to_pylist() == expected == result.column('second').to_pylist() - assert result.schema.field('first').nullable and result.schema.field('second').nullable + assert result.schema == pa.schema([pa.field('first', target), pa.field('second', target), id_field]) builder = table.copy({'read.native.enabled': 'true', 'scan.native-plan.enabled': 'true'}).new_read_builder() builder.with_projection(['masked', 'id']) splits = builder.new_scan().plan().splits() diff --git a/paimon-python/pypaimon/tests/native_system_columns_test.py b/paimon-python/pypaimon/tests/native_system_columns_test.py index 8ba62eab7a2b..03dbf5d4e24e 100644 --- a/paimon-python/pypaimon/tests/native_system_columns_test.py +++ b/paimon-python/pypaimon/tests/native_system_columns_test.py @@ -24,6 +24,8 @@ import pytest from pypaimon import CatalogFactory, Schema +from pypaimon.casting.schema_evolution import read_arrow_schema +from pypaimon.table.special_fields import SpecialFields pytestmark = pytest.mark.native_plan _SCHEMA = pa.schema([('id', pa.int32()), ('v', pa.int32()), ('w', pa.int32())]) @@ -86,13 +88,16 @@ def _read(table, native, projection, predicate=None, limit=None): @pytest.mark.parametrize('deletion_vectors', [False, True]) def test_partial_column_updates_keep_latest_metadata_provider(tmp_path, native, native_write, deletion_vectors): table = _table(tmp_path, True, native_write, deletion_vectors) + assert not SpecialFields.ROW_ID.type.nullable + assert not SpecialFields.SEQUENCE_NUMBER.type.nullable _update(table, 'v', 0, 77) _update(table, 'w', 1, 88) for projection in (['_SEQUENCE_NUMBER'], ['id', '_SEQUENCE_NUMBER'], ['_SEQUENCE_NUMBER', 'w', '_ROW_ID', 'v', 'id']): result = _read(table, native, projection) assert result.column_names == projection - assert not result.schema.field('_SEQUENCE_NUMBER').nullable + expected_schema = read_arrow_schema(table.new_read_builder().with_projection(projection).read_type()) + assert result.schema == expected_schema assert result['_SEQUENCE_NUMBER'].null_count == 0 if 'id' in projection: result = result.sort_by('id') diff --git a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py index 9257c8fd4de4..2b8dac370ce7 100644 --- a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py +++ b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py @@ -30,6 +30,7 @@ from pypaimon import CatalogFactory, Schema from pypaimon.ray import read_by_row_id +from pypaimon.schema.data_types import PyarrowFieldParser class RayReadByRowIdTest(unittest.TestCase): @@ -530,7 +531,26 @@ def test_empty_result_schema_matches_nonempty_read(self): target, pa.table({"_ROW_ID": pa.array([], pa.int64())}), self.catalog_options, projection=proj) self.assertTrue(real_schema.equals(empty.schema().base_schema)) - self.assertFalse(empty.schema().base_schema.field("_ROW_ID").nullable) + + def test_empty_result_keeps_nested_read_schema_and_projection_order(self): + schema = pa.schema([ + pa.field("id", pa.int32(), nullable=False), + pa.field("details", pa.struct([pa.field("value", pa.int64(), nullable=False)]), nullable=False), + ]) + target = self._create(schema=schema) + self._write(target, pa.Table.from_pylist([{"id": 1, "details": {"value": 7}}], schema=schema)) + projection = ["details", "id"] + actual = self._read(target, projection + ["_ROW_ID"]) + self.assertEqual(actual.to_pylist(), [{"details": {"value": 7}, "id": 1, "_ROW_ID": 0}]) + for name in [target, self._create(schema=schema)]: + with self.subTest(target=name): + empty = read_by_row_id( + name, pa.table({"_ROW_ID": pa.array([], pa.int64())}), + self.catalog_options, projection=projection) + self.assertEqual(empty.count(), 0) + self.assertEqual(empty.schema().base_schema, actual.schema) + logical_schema = PyarrowFieldParser.from_paimon_schema(self.catalog.get_table(name).fields) + self.assertEqual(logical_schema, schema) def test_custom_row_id_col(self): target = self._create() diff --git a/paimon-python/pypaimon/tests/reader_primary_key_test.py b/paimon-python/pypaimon/tests/reader_primary_key_test.py index 4a1d59043b79..438e57ac649a 100644 --- a/paimon-python/pypaimon/tests/reader_primary_key_test.py +++ b/paimon-python/pypaimon/tests/reader_primary_key_test.py @@ -24,6 +24,7 @@ import pyarrow as pa from pypaimon import CatalogFactory, Schema +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.common.options.core_options import CoreOptions @@ -48,7 +49,7 @@ def setUpClass(cls): 'item_id': [1001, 1002, 1003, 1004, 1005, 1007, 1008], 'behavior': ['a', 'b-new', 'c', None, 'e', 'g', 'h'], 'dt': ['p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2'], - }, schema=cls.pa_schema) + }, schema=read_arrow_schema(Schema.from_pyarrow_schema(cls.pa_schema).fields)) @classmethod def tearDownClass(cls): @@ -276,7 +277,7 @@ def test_pk_multi_write_once_commit(self): 'item_id': [1001, 1002, 1003, 1004, 1005, 1007, 1008], 'behavior': ['a', 'b-new', 'c', None, 'e', 'g', 'h'], 'dt': ['p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2'], - }, schema=self.pa_schema) + }, schema=read_arrow_schema(table.fields)) self.assertEqual(actual, expected) def test_nullable_primary_key(self): @@ -481,7 +482,7 @@ def test_incremental_timestamp(self): 'item_id': [1001, 1002, 1002, 1003, 1004, 1005, 1007, 1008], 'behavior': ['a', 'b', 'b-new', 'c', None, 'e', 'g', 'h'], 'dt': ['p1', 'p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2'], - }, schema=self.pa_schema) + }, schema=read_arrow_schema(table.fields)) self.assertEqual(expected, actual) # test 3 table = table.copy({CoreOptions.INCREMENTAL_BETWEEN_TIMESTAMP.key(): str(t1) + ',' + str(t2)}) @@ -492,7 +493,7 @@ def test_incremental_timestamp(self): "item_id": [1002, 1005, 1007, 1008], "behavior": ["b-new", "e", "g", "h"], "dt": ["p1", "p2", "p1", "p2"] - }, schema=self.pa_schema) + }, schema=read_arrow_schema(table.fields)) self.assertEqual(expected, actual) def test_incremental_read_multi_snapshots(self): @@ -530,7 +531,7 @@ def test_incremental_read_multi_snapshots(self): 'item_id': [1000 + i for i in range(11, 21)], 'behavior': [f'snap{i}' for i in range(11, 21)], 'dt': ['p1' if i % 2 == 1 else 'p2' for i in range(11, 21)], - }, schema=self.pa_schema).sort_by('user_id') + }, schema=read_arrow_schema(table.fields)).sort_by('user_id') self.assertEqual(expected, actual) def test_manifest_creation_time_timestamp(self): diff --git a/paimon-python/pypaimon/tests/rest/rest_read_write_test.py b/paimon-python/pypaimon/tests/rest/rest_read_write_test.py index eb5d4b459456..cd77d41627dc 100644 --- a/paimon-python/pypaimon/tests/rest/rest_read_write_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_read_write_test.py @@ -22,6 +22,7 @@ from pypaimon.common.options import Options from pypaimon.catalog.catalog_context import CatalogContext +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon import CatalogFactory from pypaimon.catalog.catalog_exception import TableNotExistException from pypaimon.catalog.rest.rest_catalog import RESTCatalog @@ -327,7 +328,7 @@ def test_pk_parquet_reader(self): read_builder = table.new_read_builder() actual = self._read_test_table(read_builder).sort_by('user_id') - self.assertEqual(actual, self.pk_expected) + self.assertEqual(actual, self.pk_expected.cast(read_arrow_schema(table.fields))) def test_pk_orc_reader(self): schema = Schema.from_pyarrow_schema(self.pa_schema, @@ -365,7 +366,7 @@ def test_pk_avro_reader(self): read_builder = table.new_read_builder() actual = self._read_test_table(read_builder).sort_by('user_id') - self.assertEqual(actual, expected) + self.assertEqual(actual, expected.cast(read_arrow_schema(table.fields))) def test_pk_lance_reader(self): schema = Schema.from_pyarrow_schema(self.pa_schema, @@ -385,7 +386,7 @@ def test_pk_lance_reader(self): read_builder = table.new_read_builder() actual = self._read_test_table(read_builder).sort_by('user_id') - self.assertEqual(actual, self.pk_expected) + self.assertEqual(actual, self.pk_expected.cast(read_arrow_schema(table.fields))) def test_lance_ao_reader_with_filter(self): schema = Schema.from_pyarrow_schema(self.pa_schema, partition_keys=['dt'], options={'file.format': 'lance'}) @@ -433,7 +434,7 @@ def test_pk_lance_reader_with_filter(self): self.pk_expected.slice(1, 1), # 2/b self.pk_expected.slice(5, 1) # 7/g ]) - self.assertEqual(actual, expected) + self.assertEqual(actual, expected.cast(read_arrow_schema(table.fields))) def test_pk_lance_reader_no_bucket(self): """Test Lance format with PrimaryKey table without specifying bucket.""" @@ -450,7 +451,7 @@ def test_pk_lance_reader_no_bucket(self): read_builder = table.new_read_builder() actual = self._read_test_table(read_builder).sort_by('user_id') - self.assertEqual(actual, self.pk_expected) + self.assertEqual(actual, self.pk_expected.cast(read_arrow_schema(table.fields))) def test_pk_reader_with_filter(self): schema = Schema.from_pyarrow_schema(self.pa_schema, @@ -472,7 +473,7 @@ def test_pk_reader_with_filter(self): self.pk_expected.slice(1, 1), # 2/b self.pk_expected.slice(5, 1) # 7/g ]) - self.assertEqual(actual, expected) + self.assertEqual(actual, expected.cast(read_arrow_schema(table.fields))) def test_pk_reader_with_projection(self): schema = Schema.from_pyarrow_schema(self.pa_schema, @@ -485,7 +486,7 @@ def test_pk_reader_with_projection(self): read_builder = table.new_read_builder().with_projection(['dt', 'user_id', 'behavior']) actual = self._read_test_table(read_builder).sort_by('user_id') - expected = self.pk_expected.select(['dt', 'user_id', 'behavior']) + expected = self.pk_expected.cast(read_arrow_schema(table.fields)).select(['dt', 'user_id', 'behavior']) self.assertEqual(actual, expected) def test_write_wrong_schema(self): diff --git a/paimon-python/pypaimon/tests/rest/rest_simple_test.py b/paimon-python/pypaimon/tests/rest/rest_simple_test.py index 1217661c29cc..205c5c53cd6c 100644 --- a/paimon-python/pypaimon/tests/rest/rest_simple_test.py +++ b/paimon-python/pypaimon/tests/rest/rest_simple_test.py @@ -21,6 +21,7 @@ import pytest from pypaimon import Schema +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.catalog.catalog_exception import DatabaseAlreadyExistException, TableAlreadyExistException, \ DatabaseNotExistException, TableNotExistException from pypaimon.schema.data_types import AtomicType, DataField @@ -627,7 +628,7 @@ def test_with_shard_pk_dynamic_bucket(self): splits.extend(read_builder.new_scan().with_shard(2, 3).plan().splits()) table_read = read_builder.new_read() actual = table_read.to_arrow(splits) - expected_sorted = table_sort_by(self.expected.cast(self.pk_pa_schema), 'user_id') + expected_sorted = table_sort_by(self.expected.cast(read_arrow_schema(table.fields)), 'user_id') actual_sorted = table_sort_by(actual, 'user_id') self.assertEqual(actual_sorted, expected_sorted) @@ -664,7 +665,7 @@ def test_with_shard_pk_fixed_bucket(self): 'behavior': ['b', 'c', 'a', 'd', 'e'], 'dt': ['2025-08-10', '2025-08-11', '2000-10-10', '2025-08-12', '2025-08-13'] } - expected = pa.Table.from_pydict(data_expected, schema=table_pa_schema) + expected = pa.Table.from_pydict(data_expected, schema=read_arrow_schema(table.fields)) self.assertEqual(actual.combine_chunks(), expected.combine_chunks()) def test_with_shard_uniform_division(self): diff --git a/paimon-python/pypaimon/tests/search_result_metadata_test.py b/paimon-python/pypaimon/tests/search_result_metadata_test.py index 36fb34f413c3..3e1090607ca6 100644 --- a/paimon-python/pypaimon/tests/search_result_metadata_test.py +++ b/paimon-python/pypaimon/tests/search_result_metadata_test.py @@ -105,6 +105,23 @@ def test_duplicate_projections_keep_scores(docs, batch, column): assert (with_lookup[0] if batch else with_lookup).schema == (actual[0] if batch else actual).schema +@pytest.mark.parametrize("batch", [False, True]) +@pytest.mark.parametrize("empty", [False, True]) +@pytest.mark.parametrize("projection", [[], ["_ROW_ID"], ["_ROW_ID", "_ROW_ID"]]) +def test_metadata_only_matches_lookup_schema_and_values(docs, batch, empty, projection): + kwargs = {"pre_filter": "id < 0"} if empty else {} + with patch.object(ScanQuery, "_read_global_index_result", side_effect=AssertionError("final lookup")): + actual = query(docs, batch, **kwargs).select(projection).with_score().order_by_score().to_arrow() + expected = (query(docs, batch, **kwargs).select(projection).where("id >= 0") + .with_score().order_by_score().to_arrow()) + actual, expected = (actual, expected) if batch else ([actual], [expected]) + assert len(actual) == len(expected) == (2 if batch else 1) + for fast, lookup in zip(actual, expected): + assert fast.num_rows == (0 if empty else 4) + assert fast.schema == lookup.schema + assert fast.equals(lookup) + + @pytest.mark.parametrize("batch", [False, True]) def test_post_filter_and_deleted_rows_keep_score_alignment(docs, batch): docs.delete("id = 1") diff --git a/paimon-python/pypaimon/tests/serializable_test.py b/paimon-python/pypaimon/tests/serializable_test.py index 0222c0a6b028..5d7c1b00bb5d 100644 --- a/paimon-python/pypaimon/tests/serializable_test.py +++ b/paimon-python/pypaimon/tests/serializable_test.py @@ -24,6 +24,7 @@ import pyarrow as pa from pypaimon import CatalogFactory +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon import Schema @@ -54,7 +55,7 @@ def setUpClass(cls): 'item_id': [1001, 1002, 1003, 1004, 1005, 1007, 1008], 'behavior': ['a', 'b-new', 'c', None, 'e', 'g', 'h'], 'dt': ['p1', 'p1', 'p2', 'p1', 'p2', 'p1', 'p2'], - }, schema=cls.pk_pa_schema) + }, schema=read_arrow_schema(Schema.from_pyarrow_schema(cls.pk_pa_schema).fields)) @classmethod def tearDownClass(cls): diff --git a/paimon-python/pypaimon/tests/table_commit_test.py b/paimon-python/pypaimon/tests/table_commit_test.py index 8a213fd10d2c..2eaa53a4fde1 100644 --- a/paimon-python/pypaimon/tests/table_commit_test.py +++ b/paimon-python/pypaimon/tests/table_commit_test.py @@ -96,6 +96,7 @@ def fail_commit(*args, **kwargs): def test_empty_append_snapshot_is_opt_in_and_can_be_tagged(self): import pyarrow as pa import pypaimon.multimodal as pmm + from pypaimon.casting.schema_evolution import read_arrow_schema with TemporaryDirectory(prefix="paimon-empty-commit-") as warehouse: connection = pmm.connect(options={"warehouse": warehouse}) @@ -126,7 +127,8 @@ def commit_empty(): table.raw_table.create_tag("empty") tagged = table.scan(tag_name="empty").to_arrow() self.assertEqual(0, tagged.num_rows) - self.assertEqual(schema, tagged.schema) + self.assertFalse(table.raw_table.fields[0].type.nullable) + self.assertEqual(read_arrow_schema(table.raw_table.fields), tagged.schema) table.add([{"feature": "state_imu_body"}]) table.add(empty) diff --git a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py index 63ebe839e186..1542adb10457 100644 --- a/paimon-python/pypaimon/tests/table_upsert_by_key_test.py +++ b/paimon-python/pypaimon/tests/table_upsert_by_key_test.py @@ -24,6 +24,7 @@ import pyarrow.parquet as pq import pytest +from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.read.table_read import TableRead from pypaimon.table.special_fields import SpecialFields from pypaimon.write.table_update_by_row_id import _RowIdUpdateFileWriter @@ -1270,7 +1271,7 @@ def test_not_null_update_across_read_batches(self): schema.field('score'), pa.array([0, 1, 99, 3], type=pa.int32()), ) - self.assertTrue(self._read_all(table).equals(expected)) + self.assertTrue(self._read_all(table).equals(expected.cast(read_arrow_schema(table.fields)))) files = [file for message in messages for file in message.new_files] self.assertEqual(len(files), 1) self.assertFalse(pq.read_schema(files[0].file_path).field('score').nullable) diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py b/paimon-python/pypaimon/tests/write/table_write_test.py index 8aa93a53ad13..667fd4599f4b 100644 --- a/paimon-python/pypaimon/tests/write/table_write_test.py +++ b/paimon-python/pypaimon/tests/write/table_write_test.py @@ -25,6 +25,7 @@ from unittest.mock import Mock, patch from pypaimon import CatalogFactory, Schema +from pypaimon.casting.schema_evolution import read_arrow_schema import pyarrow as pa import pytest from parameterized import parameterized @@ -102,7 +103,9 @@ def arrow_schema(string_type): commit.close() actual = self._read_sorted(table, [('id', 'ascending')]) self.assertEqual(actual.to_pylist(), expected) - self.assertEqual(actual.schema.remove_metadata(), arrow_schema(pa.string())) + self.assertFalse(table.fields[0].type.nullable) + expected_schema = read_arrow_schema(Schema.from_pyarrow_schema(arrow_schema(pa.string())).fields) + self.assertEqual(actual.schema.remove_metadata(), expected_schema) @classmethod def setUpClass(cls): @@ -564,7 +567,7 @@ def test_write_row_fixed_bucket_primary_key_table(self): 'item_id': [1001, 1002], 'behavior': ['a', 'b'], 'dt': ['p1', 'p2'], - }, schema=self.pk_pa_schema) + }, schema=read_arrow_schema(table.fields)) sort_keys = [('user_id', 'ascending'), ('dt', 'ascending')] self.assertEqual( expected.sort_by(sort_keys), self._read_sorted(table, sort_keys)) @@ -594,7 +597,7 @@ def test_write_row_dynamic_bucket_primary_key_table(self): 'item_id': [1001, 1002], 'behavior': ['a', 'b'], 'dt': ['p1', 'p2'], - }, schema=self.pk_pa_schema) + }, schema=read_arrow_schema(table.fields)) sort_keys = [('user_id', 'ascending'), ('dt', 'ascending')] self.assertEqual( expected.sort_by(sort_keys), self._read_sorted(table, sort_keys)) @@ -774,7 +777,7 @@ def test_multi_prepare_commit_pk(self): table_read = read_builder.new_read() splits = read_builder.new_scan().plan().splits() actual = table_read.to_arrow(splits).sort_by('user_id') - self.assertEqual(self.pk_expected, actual) + self.assertEqual(self.pk_expected.cast(read_arrow_schema(table.fields)), actual) def test_postpone_read_write(self): table = self._create_postpone_table( @@ -828,7 +831,7 @@ def test_postpone_read_write(self): table_read = read_builder.new_read() splits = read_builder.new_scan().plan().splits() actual = table_read.to_arrow(splits) - self.assertEqual(expect, actual) + self.assertEqual(expect.cast(read_arrow_schema(table.fields)), actual) def test_postpone_file_store_write_validates_runtime_bucket_count(self): from pypaimon.write.file_store_write import ( @@ -1889,7 +1892,7 @@ def test_dynamic_bucket_write(self): actual = table_read.to_arrow(splits) sort_keys = [('user_id', 'ascending'), ('dt', 'ascending')] self.assertEqual( - self.pk_expected.sort_by(sort_keys), + self.pk_expected.cast(read_arrow_schema(table.fields)).sort_by(sort_keys), actual.sort_by(sort_keys), ) From 5d0d6135372c932ac190950653561954d7ba0e62 Mon Sep 17 00:00:00 2001 From: Yann Date: Thu, 15 Oct 2026 11:56:15 +0800 Subject: [PATCH 2/4] test(python): verify metadata search preserves core row-id nullability Exercise the core schema boundary without requiring a new native wheel. The test rejects the previous hard-coded non-nullable metadata field. Co-Authored-By: Codex AI-Model: gpt-6 Co-Authored-By: Codex AI-Contributed/Feature: 0/0 AI-Contributed/UT: 11/11 --- .../pypaimon/tests/search_result_metadata_test.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/paimon-python/pypaimon/tests/search_result_metadata_test.py b/paimon-python/pypaimon/tests/search_result_metadata_test.py index 3e1090607ca6..73d2aaef0bf4 100644 --- a/paimon-python/pypaimon/tests/search_result_metadata_test.py +++ b/paimon-python/pypaimon/tests/search_result_metadata_test.py @@ -21,6 +21,7 @@ import pytest import pypaimon.multimodal as pm +from pypaimon.globalindex.vector_search_result import DictBasedScoredIndexResult from pypaimon.multimodal.query import ScanQuery @@ -122,6 +123,16 @@ def test_metadata_only_matches_lookup_schema_and_values(docs, batch, empty, proj assert fast.equals(lookup) +def test_metadata_only_preserves_core_row_id_nullability(docs): + schema = pa.schema([pa.field("_ROW_ID", pa.int64(), nullable=True)]) + search = query(docs).select(["_ROW_ID"]).with_score()._for_execution() + result = DictBasedScoredIndexResult({1: 1.}) + with patch("pypaimon.multimodal.query.read_arrow_schema", return_value=schema, create=True): + actual = search._read_global_index_result(result) + assert actual.schema == schema.append(pa.field("_score", pa.float64())) + assert actual.to_pylist() == [{"_ROW_ID": 1, "_score": 1.}] + + @pytest.mark.parametrize("batch", [False, True]) def test_post_filter_and_deleted_rows_keep_score_alignment(docs, batch): docs.delete("id = 1") From f85b254b4b54241dea4a2f5dac22a28429b49043 Mon Sep 17 00:00:00 2001 From: Yann Date: Thu, 15 Oct 2026 13:56:43 +0800 Subject: [PATCH 3/4] test(python): accept native list nullability error capitalization Match the Arrow ListArray Non-nullable diagnostic while preserving the ValueError and invalid-NULL write checks. Co-Authored-By: Codex AI-Model: gpt-6 Co-Authored-By: Codex AI-Contributed/Feature: 0/0 AI-Contributed/UT: 2/2 --- paimon-python/pypaimon/tests/blob_table_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/paimon-python/pypaimon/tests/blob_table_test.py b/paimon-python/pypaimon/tests/blob_table_test.py index 8818e85a071f..1cf27b82a43b 100755 --- a/paimon-python/pypaimon/tests/blob_table_test.py +++ b/paimon-python/pypaimon/tests/blob_table_test.py @@ -1813,7 +1813,7 @@ def test_array_blob_element_not_null(self): 'payloads': pa.array([[b'bad', None]], type=nullable_array_type), }, schema=incompatible_schema) incompatible_writer = table.new_batch_write_builder().new_write() - with self.assertRaisesRegex(ValueError, "Input schema isn't consistent|non-nullable"): + with self.assertRaisesRegex(ValueError, "Input schema isn't consistent|[Nn]on-nullable"): incompatible_writer.write_arrow(incompatible_data) incompatible_writer.abort() From 3dd3b0c4237a158f8c0a3ec63c107f651fe65d75 Mon Sep 17 00:00:00 2001 From: Yann Date: Thu, 15 Oct 2026 14:27:05 +0800 Subject: [PATCH 4/4] test(python): reuse existing schema regression coverage Reuse the existing Ray and search schema regressions and remove redundant schema assertions. Normalize the BinaryRow read fixture after setup writes so individual tests keep their original strict comparisons. Co-Authored-By: Codex AI-Model: gpt-6 Co-Authored-By: Codex AI-Contributed/Feature: 0/0 AI-Contributed/UT: 104/104 --- .../pypaimon/tests/binary_row_test.py | 24 ++++++++-------- .../pypaimon/tests/blob_table_test.py | 5 ---- .../pypaimon/tests/data_evolution_test.py | 4 --- .../tests/map_shared_shredding_write_test.py | 6 +--- .../tests/native_cast_completion_test.py | 8 ++---- .../tests/native_system_columns_test.py | 6 ---- .../pypaimon/tests/ray_read_by_row_id_test.py | 21 -------------- .../tests/search_result_metadata_test.py | 28 ------------------- .../pypaimon/tests/table_commit_test.py | 1 - .../pypaimon/tests/write/table_write_test.py | 1 - 10 files changed, 15 insertions(+), 89 deletions(-) diff --git a/paimon-python/pypaimon/tests/binary_row_test.py b/paimon-python/pypaimon/tests/binary_row_test.py index 5d4efa18095c..940240ab6cac 100644 --- a/paimon-python/pypaimon/tests/binary_row_test.py +++ b/paimon-python/pypaimon/tests/binary_row_test.py @@ -79,7 +79,7 @@ def setUpClass(cls): write.close() commit.close() - cls.expected = cls.data.cast(read_arrow_schema(append_table.fields)) + cls.data = cls.data.cast(read_arrow_schema(append_table.fields)) @classmethod def tearDownClass(cls): @@ -93,7 +93,7 @@ def test_not_equal_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.not_equal('f2', 6) # test stats filter when filtering ManifestEntry splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.slice(1, 4) + expected = self.data.slice(1, 4) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) @@ -107,7 +107,7 @@ def test_less_than_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.less_than('f2', 8) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.slice(0, 2) + expected = self.data.slice(0, 2) self.assertEqual(actual, expected) self.assertEqual(len(expected), len(splits)) # test stats filter when filtering ManifestEntry @@ -117,7 +117,7 @@ def test_is_null_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_null('f1') # value_stats_cols=None splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.slice(4, 1) + expected = self.data.slice(4, 1) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) @@ -143,7 +143,7 @@ def test_is_not_null_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_not_null('f1') splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.slice(0, 4) + expected = self.data.slice(0, 4) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) @@ -155,7 +155,7 @@ def test_is_in_append(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_in('f2', [6, 8]) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.take([0, 2]) + expected = self.data.take([0, 2]) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) @@ -165,7 +165,7 @@ def test_equal_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.equal('f2', 6) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.slice(0, 1) + expected = self.data.slice(0, 1) self.assertEqual(expected, actual) self.assertEqual(len(splits), len(expected)) # test partition filter when filtering ManifestEntry @@ -176,7 +176,7 @@ def test_not_equal_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.not_equal('f2', 6) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.slice(1, 4) + expected = self.data.slice(1, 4) self.assertEqual(actual, expected) self.assertEqual(len(splits), len(expected)) # test partition filter when filtering ManifestEntry @@ -187,7 +187,7 @@ def test_less_than_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.less_than('f0', 3) splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.slice(0, 2) + expected = self.data.slice(0, 2) self.assertEqual(expected, actual) self.assertEqual(len(expected), len(splits)) # test key stats filter when filtering ManifestEntry @@ -197,7 +197,7 @@ def test_is_null_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_null('f1') splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.slice(4, 1) + expected = self.data.slice(4, 1) self.assertEqual(actual, expected) @pytest.mark.python_plan @@ -207,7 +207,7 @@ def test_is_not_null_pk(self): predicate_builder = read_builder.new_predicate_builder() predicate = predicate_builder.is_not_null('f1') splits, actual = self._read_result(read_builder.with_filter(predicate)) - expected = self.expected.slice(0, 4) + expected = self.data.slice(0, 4) self.assertEqual(actual, expected) @pytest.mark.python_plan @@ -218,7 +218,7 @@ def test_is_in_pk(self): predicate = predicate_builder.is_in('f0', [1, 5]) splits, actual = self._read_result(read_builder.with_filter(predicate)) # expected rows: indices [0, 3] - expected = self.expected.take([0, 4]) + expected = self.data.take([0, 4]) self.assertEqual(actual, expected) self.assertEqual(len(splits), len(expected)) # test key stats filter when filtering ManifestEntry diff --git a/paimon-python/pypaimon/tests/blob_table_test.py b/paimon-python/pypaimon/tests/blob_table_test.py index 1cf27b82a43b..bf6612bf15cb 100755 --- a/paimon-python/pypaimon/tests/blob_table_test.py +++ b/paimon-python/pypaimon/tests/blob_table_test.py @@ -26,7 +26,6 @@ import pytest from pypaimon import CatalogFactory, Schema -from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.schema.schema_change import SchemaChange from pypaimon.table.file_store_table import FileStoreTable from pypaimon.table.row.blob import Blob @@ -1717,8 +1716,6 @@ def test_map_blob_value_not_null(self): read_builder = table.new_read_builder() result = read_builder.new_read().to_arrow( read_builder.new_scan().plan().splits()) - self.assertFalse(table.field_dict['payloads'].type.value.nullable) - self.assertEqual(result.schema, read_arrow_schema(table.fields)) self.assertEqual( { row['id']: None if row['payloads'] is None else dict(row['payloads']) @@ -1789,8 +1786,6 @@ def test_array_blob_element_not_null(self): read_builder = table.new_read_builder() result = read_builder.new_read().to_arrow( read_builder.new_scan().plan().splits()) - self.assertFalse(table.field_dict['payloads'].type.element.nullable) - self.assertEqual(result.schema, read_arrow_schema(table.fields)) self.assertEqual( { row['id']: row['payloads'] diff --git a/paimon-python/pypaimon/tests/data_evolution_test.py b/paimon-python/pypaimon/tests/data_evolution_test.py index 38d7c131c31a..024bf9cc7d70 100644 --- a/paimon-python/pypaimon/tests/data_evolution_test.py +++ b/paimon-python/pypaimon/tests/data_evolution_test.py @@ -32,7 +32,6 @@ from pypaimon.common.predicate import Predicate from pypaimon.manifest.manifest_list_manager import ManifestListManager from pypaimon.table.row.offset_row import OffsetRow -from pypaimon.table.special_fields import SpecialFields def _filter_batch_arrow(batch, predicate): @@ -1051,9 +1050,6 @@ def test_multiple_appends_different_first_row_ids(self): table_scan = read_builder.new_scan() table_read = read_builder.new_read() actual_with_meta = table_read.to_arrow(table_scan.plan().splits()) - self.assertFalse(SpecialFields.ROW_ID.type.nullable) - self.assertFalse(SpecialFields.SEQUENCE_NUMBER.type.nullable) - self.assertEqual(actual_with_meta.schema, read_arrow_schema(read_builder.read_type())) self.assertEqual(actual_with_meta.column('_ROW_ID').null_count, 0) self.assertEqual(actual_with_meta.column('_SEQUENCE_NUMBER').null_count, 0) diff --git a/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py b/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py index 13d13b9ff50b..41954316807b 100644 --- a/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py +++ b/paimon-python/pypaimon/tests/map_shared_shredding_write_test.py @@ -26,7 +26,6 @@ import pyarrow.parquet as pq from pypaimon import CatalogFactory, Schema -from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.data.map_shared_shredding import ( is_shared_shredding, parse_shared_shredding_metadata, @@ -170,11 +169,8 @@ def test_non_nullable_values(self): self.assertTrue(field.type['__col_1'].nullable) reader = table.new_read_builder() result = reader.new_read().to_arrow(reader.new_scan().plan().splits()) - self.assertFalse(table.field_dict['metrics'].type.value.nullable) - expected_schema = read_arrow_schema(table.fields) - self.assertEqual(result.schema, expected_schema) # Arrow 6 cannot convert non-nullable MAP values to scalars. - read_value_type = expected_schema.field('metrics').type.item_type + read_value_type = result.schema.field('metrics').type.item_type actual = result.column('metrics').cast(pa.map_(pa.string(), read_value_type)) self.assertEqual(expected, actual.to_pylist()) diff --git a/paimon-python/pypaimon/tests/native_cast_completion_test.py b/paimon-python/pypaimon/tests/native_cast_completion_test.py index e7b05fd78dcd..cc9b6fdbd3c6 100644 --- a/paimon-python/pypaimon/tests/native_cast_completion_test.py +++ b/paimon-python/pypaimon/tests/native_cast_completion_test.py @@ -32,7 +32,6 @@ from pypaimon import Schema from pypaimon.api.api_response import AuthTableQueryResponse -from pypaimon.casting.schema_evolution import read_arrow_schema from pypaimon.common.predicate_builder import PredicateBuilder from pypaimon.schema.data_types import DataTypeParser, PyarrowFieldParser from pypaimon.schema.schema_change import SchemaChange @@ -194,15 +193,12 @@ def test_not_null_mask_adapts_only_output_schema(rest_catalog, table_mode, mask) expected = [None, None] if mask == 'null' else [Decimal('1.00'), None] result = _read(table, ['masked', 'id']).sort_by([('id', 'ascending')]) assert result.column('masked').to_pylist() == expected - assert not table.field_dict['masked'].type.nullable - assert not table.field_dict['id'].type.nullable - id_field = read_arrow_schema(table.fields).field('id') - assert result.schema == pa.schema([pa.field('masked', target), id_field]) + assert result.schema.field('masked').nullable assert _read(table, ['masked'], limit=0).schema.field('masked').nullable projection = {'first': 'masked', 'second': 'masked', 'id': 'id'} result = _read(table, projection).sort_by([('id', 'ascending')]) assert result.column('first').to_pylist() == expected == result.column('second').to_pylist() - assert result.schema == pa.schema([pa.field('first', target), pa.field('second', target), id_field]) + assert result.schema.field('first').nullable and result.schema.field('second').nullable builder = table.copy({'read.native.enabled': 'true', 'scan.native-plan.enabled': 'true'}).new_read_builder() builder.with_projection(['masked', 'id']) splits = builder.new_scan().plan().splits() diff --git a/paimon-python/pypaimon/tests/native_system_columns_test.py b/paimon-python/pypaimon/tests/native_system_columns_test.py index 03dbf5d4e24e..9c6fceb4e6fd 100644 --- a/paimon-python/pypaimon/tests/native_system_columns_test.py +++ b/paimon-python/pypaimon/tests/native_system_columns_test.py @@ -24,8 +24,6 @@ import pytest from pypaimon import CatalogFactory, Schema -from pypaimon.casting.schema_evolution import read_arrow_schema -from pypaimon.table.special_fields import SpecialFields pytestmark = pytest.mark.native_plan _SCHEMA = pa.schema([('id', pa.int32()), ('v', pa.int32()), ('w', pa.int32())]) @@ -88,16 +86,12 @@ def _read(table, native, projection, predicate=None, limit=None): @pytest.mark.parametrize('deletion_vectors', [False, True]) def test_partial_column_updates_keep_latest_metadata_provider(tmp_path, native, native_write, deletion_vectors): table = _table(tmp_path, True, native_write, deletion_vectors) - assert not SpecialFields.ROW_ID.type.nullable - assert not SpecialFields.SEQUENCE_NUMBER.type.nullable _update(table, 'v', 0, 77) _update(table, 'w', 1, 88) for projection in (['_SEQUENCE_NUMBER'], ['id', '_SEQUENCE_NUMBER'], ['_SEQUENCE_NUMBER', 'w', '_ROW_ID', 'v', 'id']): result = _read(table, native, projection) assert result.column_names == projection - expected_schema = read_arrow_schema(table.new_read_builder().with_projection(projection).read_type()) - assert result.schema == expected_schema assert result['_SEQUENCE_NUMBER'].null_count == 0 if 'id' in projection: result = result.sort_by('id') diff --git a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py index 2b8dac370ce7..1627acc0c568 100644 --- a/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py +++ b/paimon-python/pypaimon/tests/ray_read_by_row_id_test.py @@ -30,7 +30,6 @@ from pypaimon import CatalogFactory, Schema from pypaimon.ray import read_by_row_id -from pypaimon.schema.data_types import PyarrowFieldParser class RayReadByRowIdTest(unittest.TestCase): @@ -532,26 +531,6 @@ def test_empty_result_schema_matches_nonempty_read(self): self.catalog_options, projection=proj) self.assertTrue(real_schema.equals(empty.schema().base_schema)) - def test_empty_result_keeps_nested_read_schema_and_projection_order(self): - schema = pa.schema([ - pa.field("id", pa.int32(), nullable=False), - pa.field("details", pa.struct([pa.field("value", pa.int64(), nullable=False)]), nullable=False), - ]) - target = self._create(schema=schema) - self._write(target, pa.Table.from_pylist([{"id": 1, "details": {"value": 7}}], schema=schema)) - projection = ["details", "id"] - actual = self._read(target, projection + ["_ROW_ID"]) - self.assertEqual(actual.to_pylist(), [{"details": {"value": 7}, "id": 1, "_ROW_ID": 0}]) - for name in [target, self._create(schema=schema)]: - with self.subTest(target=name): - empty = read_by_row_id( - name, pa.table({"_ROW_ID": pa.array([], pa.int64())}), - self.catalog_options, projection=projection) - self.assertEqual(empty.count(), 0) - self.assertEqual(empty.schema().base_schema, actual.schema) - logical_schema = PyarrowFieldParser.from_paimon_schema(self.catalog.get_table(name).fields) - self.assertEqual(logical_schema, schema) - def test_custom_row_id_col(self): target = self._create() self._write(target, pa.Table.from_pydict( diff --git a/paimon-python/pypaimon/tests/search_result_metadata_test.py b/paimon-python/pypaimon/tests/search_result_metadata_test.py index 73d2aaef0bf4..36fb34f413c3 100644 --- a/paimon-python/pypaimon/tests/search_result_metadata_test.py +++ b/paimon-python/pypaimon/tests/search_result_metadata_test.py @@ -21,7 +21,6 @@ import pytest import pypaimon.multimodal as pm -from pypaimon.globalindex.vector_search_result import DictBasedScoredIndexResult from pypaimon.multimodal.query import ScanQuery @@ -106,33 +105,6 @@ def test_duplicate_projections_keep_scores(docs, batch, column): assert (with_lookup[0] if batch else with_lookup).schema == (actual[0] if batch else actual).schema -@pytest.mark.parametrize("batch", [False, True]) -@pytest.mark.parametrize("empty", [False, True]) -@pytest.mark.parametrize("projection", [[], ["_ROW_ID"], ["_ROW_ID", "_ROW_ID"]]) -def test_metadata_only_matches_lookup_schema_and_values(docs, batch, empty, projection): - kwargs = {"pre_filter": "id < 0"} if empty else {} - with patch.object(ScanQuery, "_read_global_index_result", side_effect=AssertionError("final lookup")): - actual = query(docs, batch, **kwargs).select(projection).with_score().order_by_score().to_arrow() - expected = (query(docs, batch, **kwargs).select(projection).where("id >= 0") - .with_score().order_by_score().to_arrow()) - actual, expected = (actual, expected) if batch else ([actual], [expected]) - assert len(actual) == len(expected) == (2 if batch else 1) - for fast, lookup in zip(actual, expected): - assert fast.num_rows == (0 if empty else 4) - assert fast.schema == lookup.schema - assert fast.equals(lookup) - - -def test_metadata_only_preserves_core_row_id_nullability(docs): - schema = pa.schema([pa.field("_ROW_ID", pa.int64(), nullable=True)]) - search = query(docs).select(["_ROW_ID"]).with_score()._for_execution() - result = DictBasedScoredIndexResult({1: 1.}) - with patch("pypaimon.multimodal.query.read_arrow_schema", return_value=schema, create=True): - actual = search._read_global_index_result(result) - assert actual.schema == schema.append(pa.field("_score", pa.float64())) - assert actual.to_pylist() == [{"_ROW_ID": 1, "_score": 1.}] - - @pytest.mark.parametrize("batch", [False, True]) def test_post_filter_and_deleted_rows_keep_score_alignment(docs, batch): docs.delete("id = 1") diff --git a/paimon-python/pypaimon/tests/table_commit_test.py b/paimon-python/pypaimon/tests/table_commit_test.py index 2eaa53a4fde1..bc6628cc56eb 100644 --- a/paimon-python/pypaimon/tests/table_commit_test.py +++ b/paimon-python/pypaimon/tests/table_commit_test.py @@ -127,7 +127,6 @@ def commit_empty(): table.raw_table.create_tag("empty") tagged = table.scan(tag_name="empty").to_arrow() self.assertEqual(0, tagged.num_rows) - self.assertFalse(table.raw_table.fields[0].type.nullable) self.assertEqual(read_arrow_schema(table.raw_table.fields), tagged.schema) table.add([{"feature": "state_imu_body"}]) diff --git a/paimon-python/pypaimon/tests/write/table_write_test.py b/paimon-python/pypaimon/tests/write/table_write_test.py index 667fd4599f4b..576947c526bb 100644 --- a/paimon-python/pypaimon/tests/write/table_write_test.py +++ b/paimon-python/pypaimon/tests/write/table_write_test.py @@ -103,7 +103,6 @@ def arrow_schema(string_type): commit.close() actual = self._read_sorted(table, [('id', 'ascending')]) self.assertEqual(actual.to_pylist(), expected) - self.assertFalse(table.fields[0].type.nullable) expected_schema = read_arrow_schema(Schema.from_pyarrow_schema(arrow_schema(pa.string())).fields) self.assertEqual(actual.schema.remove_metadata(), expected_schema)